[refactor][ml] Replace cache eviction algorithm with centralized removal queue and job

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AbstractEntryImpl.java

@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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
+ *
+ *   http://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 org.apache.bookkeeper.mledger.impl;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.Recycler;
+import io.netty.util.ReferenceCounted;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.util.AbstractCASReferenceCounted;
+
+public abstract class AbstractEntryImpl<T extends AbstractEntryImpl<T>> extends AbstractCASReferenceCounted
+        implements Entry, Comparable<T> {
+    protected final Recycler.Handle<T> recyclerHandle;
+    protected long timestamp;
+    protected long ledgerId;
+    protected long entryId;
+    private ByteBuf data;
+    private int length = -1;
+    private Position position;
+    private Runnable onDeallocate;
+
+    public AbstractEntryImpl(Recycler.Handle<T> recyclerHandle) {
+        this.recyclerHandle = recyclerHandle;
+    }
+
+    public long getTimestamp() {
+        return timestamp;
+    }
+
+    @Override
+    public ByteBuf getDataBuffer() {
+        return data;
+    }
+
+    protected void setDataBuffer(ByteBuf data) {
+        this.data = data;
+        this.length = data.readableBytes();
+    }
+
+    @Override
+    public byte[] getData() {
+        ByteBuf data = getDataBuffer().duplicate();
+        byte[] array = new byte[data.readableBytes()];
+        data.getBytes(data.readerIndex(), array);
+        return array;
+    }
+
+    // Only for test
+
+    @Override
+    public byte[] getDataAndRelease() {
+        byte[] array = getData();
+        release();
+        return array;
+    }
+
+    @Override
+    public int getLength() {
+        if (length == -1) {
+            throw new IllegalStateException("Entry has no length. Call setDataBuffer to set the data buffer first.");
+        }
+        return length;
+    }
+
+    @Override
+    public Position getPosition() {
+        if (position == null) {
+            position = PositionFactory.create(ledgerId, entryId);
+        }
+        return position;
+    }
+
+    @Override
+    public long getLedgerId() {
+        return ledgerId;
+    }
+
+    @Override
+    public long getEntryId() {
+        return entryId;
+    }
+
+    @Override
+    public int compareTo(T other) {
+        if (this.ledgerId != other.ledgerId) {
+            return this.ledgerId < other.ledgerId ? -1 : 1;
+        }
+
+        if (this.entryId != other.entryId) {
+            return this.entryId < other.entryId ? -1 : 1;
+        }
+
+        return 0;
+    }
+
+    @Override
+    public ReferenceCounted touch(Object hint) {
+        return this;
+    }
+
+    public void onDeallocate(Runnable r) {
+        if (this.onDeallocate == null) {
+            this.onDeallocate = r;
+        } else {
+            // this is not expected to happen
+            Runnable previous = this.onDeallocate;
+            this.onDeallocate = () -> {
+                try {
+                    previous.run();
+                } finally {
+                    r.run();
+                }
+            };
+        }
+    }
+
+    @Override
+    protected final void deallocate() {
+        beforeDeallocate();
+        // This method is called whenever the ref-count of the EntryImpl reaches 0, so that now we can recycle it
+        if (onDeallocate != null) {
+            try {
+                onDeallocate.run();
+            } finally {
+                onDeallocate = null;
+            }
+        }
+        data.release();
+        data = null;
+        length = -1;
+        timestamp = -1;
+        ledgerId = -1;
+        entryId = -1;
+        position = null;
+        beforeRecycle();
+        recyclerHandle.recycle(self());
+    }
+
+    /**
+     * This method is called just before the object is deallocated.
+     * Subclasses can override this method to run actions before the fields
+     * of the object are cleared and the object gets recycled.
+     */
+    protected void beforeDeallocate() {
+        // No-op
+    }
+
+    /**
+     * This method is called just before the object is recycled. Subclasses can override this methods to cleanup
+     * the object before it is returned to the pool.
+     */
+    protected void beforeRecycle() {
+        // No-op
+    }
+
+    @SuppressWarnings("unchecked")
+    protected T self() {
+        return (T) this;
+    }
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java

@@ -22,41 +22,47 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.util.Recycler;
-import io.netty.util.Recycler.Handle;
-import io.netty.util.ReferenceCounted;
 import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.bookkeeper.mledger.PositionFactory;
-import org.apache.bookkeeper.mledger.util.AbstractCASReferenceCounted;
-import org.apache.bookkeeper.mledger.util.RangeCache;
-
-public final class EntryImpl extends AbstractCASReferenceCounted implements Entry, Comparable<EntryImpl>,
-        RangeCache.ValueWithKeyValidation<Position> {
+import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
 
+public final class EntryImpl extends AbstractEntryImpl<EntryImpl> {
     private static final Recycler<EntryImpl> RECYCLER = new Recycler<EntryImpl>() {
         @Override
         protected EntryImpl newObject(Handle<EntryImpl> handle) {
             return new EntryImpl(handle);
         }
     };
 
-    private final Handle<EntryImpl> recyclerHandle;
-    private long timestamp;
-    private long ledgerId;
-    private long entryId;
-    private Position position;
-    ByteBuf data;
-
-    private Runnable onDeallocate;
+    public static EntryImpl create(LedgerEntry ledgerEntry, ManagedLedgerInterceptor interceptor) {
+        ManagedLedgerInterceptor.PayloadProcessorHandle processorHandle = null;
+        if (interceptor != null) {
+            ByteBuf duplicateBuffer = ledgerEntry.getEntryBuffer().retainedDuplicate();
+            processorHandle = interceptor
+                    .processPayloadBeforeEntryCache(duplicateBuffer);
+            if (processorHandle != null) {
+                ledgerEntry  = LedgerEntryImpl.create(ledgerEntry.getLedgerId(), ledgerEntry.getEntryId(),
+                        ledgerEntry.getLength(), processorHandle.getProcessedPayload());
+            } else {
+                duplicateBuffer.release();
+            }
+        }
+        EntryImpl returnEntry = create(ledgerEntry);
+        if (processorHandle != null) {
+            processorHandle.release();
+            ledgerEntry.close();
+        }
+        return returnEntry;
+    }
 
     public static EntryImpl create(LedgerEntry ledgerEntry) {
         EntryImpl entry = RECYCLER.get();
         entry.timestamp = System.nanoTime();
         entry.ledgerId = ledgerEntry.getLedgerId();
         entry.entryId = ledgerEntry.getEntryId();
-        entry.data = ledgerEntry.getEntryBuffer();
-        entry.data.retain();
+        entry.setDataBuffer(ledgerEntry.getEntryBuffer().retain());
         entry.setRefCnt(1);
         return entry;
     }
@@ -67,7 +73,7 @@ public static EntryImpl create(long ledgerId, long entryId, byte[] data) {
         entry.timestamp = System.nanoTime();
         entry.ledgerId = ledgerId;
         entry.entryId = entryId;
-        entry.data = Unpooled.wrappedBuffer(data);
+        entry.setDataBuffer(Unpooled.wrappedBuffer(data));
         entry.setRefCnt(1);
         return entry;
     }
@@ -77,8 +83,7 @@ public static EntryImpl create(long ledgerId, long entryId, ByteBuf data) {
         entry.timestamp = System.nanoTime();
         entry.ledgerId = ledgerId;
         entry.entryId = entryId;
-        entry.data = data;
-        entry.data.retain();
+        entry.setDataBuffer(data.retain());
         entry.setRefCnt(1);
         return entry;
     }
@@ -88,128 +93,22 @@ public static EntryImpl create(Position position, ByteBuf data) {
         entry.timestamp = System.nanoTime();
         entry.ledgerId = position.getLedgerId();
         entry.entryId = position.getEntryId();
-        entry.data = data;
-        entry.data.retain();
+        entry.setDataBuffer(data.retain());
         entry.setRefCnt(1);
         return entry;
     }
 
-    public static EntryImpl create(EntryImpl other) {
+    public static EntryImpl create(Entry other) {
         EntryImpl entry = RECYCLER.get();
         entry.timestamp = System.nanoTime();
-        entry.ledgerId = other.ledgerId;
-        entry.entryId = other.entryId;
-        entry.data = other.data.retainedDuplicate();
+        entry.ledgerId = other.getLedgerId();
+        entry.entryId = other.getEntryId();
+        entry.setDataBuffer(other.getDataBuffer().retainedDuplicate());
         entry.setRefCnt(1);
         return entry;
     }
 
     private EntryImpl(Recycler.Handle<EntryImpl> recyclerHandle) {
-        this.recyclerHandle = recyclerHandle;
-    }
-
-    public void onDeallocate(Runnable r) {
-        if (this.onDeallocate == null) {
-            this.onDeallocate = r;
-        } else {
-            // this is not expected to happen
-            Runnable previous = this.onDeallocate;
-            this.onDeallocate = () -> {
-                try {
-                    previous.run();
-                } finally {
-                    r.run();
-                }
-            };
-        }
-    }
-
-    public long getTimestamp() {
-        return timestamp;
-    }
-
-    @Override
-    public ByteBuf getDataBuffer() {
-        return data;
-    }
-
-    @Override
-    public byte[] getData() {
-        byte[] array = new byte[data.readableBytes()];
-        data.getBytes(data.readerIndex(), array);
-        return array;
-    }
-
-    // Only for test
-    @Override
-    public byte[] getDataAndRelease() {
-        byte[] array = getData();
-        release();
-        return array;
-    }
-
-    @Override
-    public int getLength() {
-        return data.readableBytes();
-    }
-
-    @Override
-    public Position getPosition() {
-        if (position == null) {
-            position = PositionFactory.create(ledgerId, entryId);
-        }
-        return position;
-    }
-
-    @Override
-    public long getLedgerId() {
-        return ledgerId;
-    }
-
-    @Override
-    public long getEntryId() {
-        return entryId;
-    }
-
-    @Override
-    public int compareTo(EntryImpl other) {
-        if (this.ledgerId != other.ledgerId) {
-            return this.ledgerId < other.ledgerId ? -1 : 1;
-        }
-
-        if (this.entryId != other.entryId) {
-            return this.entryId < other.entryId ? -1 : 1;
-        }
-
-        return 0;
-    }
-
-    @Override
-    public ReferenceCounted touch(Object hint) {
-        return this;
-    }
-
-    @Override
-    protected void deallocate() {
-        // This method is called whenever the ref-count of the EntryImpl reaches 0, so that now we can recycle it
-        if (onDeallocate != null) {
-            try {
-                onDeallocate.run();
-            } finally {
-                onDeallocate = null;
-            }
-        }
-        data.release();
-        data = null;
-        timestamp = -1;
-        ledgerId = -1;
-        entryId = -1;
-        position = null;
-        recyclerHandle.recycle(this);
-    }
-
-    @Override
-    public boolean matchesKey(Position key) {
-        return key.compareTo(ledgerId, entryId) == 0;
+        super(recyclerHandle);
     }
 }