perf(s3stream): increase the object overhead of StreamRecordBatch (A…

…utoMQ#1661) * perf(s3stream): increase the object overhead of `StreamRecordBatch` Signed-off-by: Ning Yu <[email protected]> * refactor: make `OBJECT_OVERHEAD` private Signed-off-by: Ning Yu <[email protected]> --------- Signed-off-by: Ning Yu <[email protected]>
Kris20030907 · Jul 29, 2024 · d2063aa · d2063aa
1 parent 116b024
commit d2063aa
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 5 deletions.
diff --git a/s3stream/src/main/java/com/automq/stream/s3/cache/LogCache.java b/s3stream/src/main/java/com/automq/stream/s3/cache/LogCache.java
@@ -38,7 +38,6 @@
 import org.slf4j.LoggerFactory;
 
 import static com.automq.stream.s3.cache.LogCache.StreamRange.NOOP_OFFSET;
-import static com.automq.stream.s3.model.StreamRecordBatch.OBJECT_OVERHEAD;
 import static com.automq.stream.utils.FutureUtil.suppress;
 
 public class LogCache {
@@ -86,7 +85,7 @@ public LogCache(long capacity, long cacheBlockMaxSize, int maxCacheBlockStreamCo
     public boolean put(StreamRecordBatch recordBatch) {
         long startTime = System.nanoTime();
         tryRealFree();
-        size.addAndGet(recordBatch.size() + OBJECT_OVERHEAD);
+        size.addAndGet(recordBatch.occupiedSize());
         readLock.lock();
         boolean full;
         try {
@@ -341,8 +340,7 @@ public boolean put(StreamRecordBatch recordBatch) {
                 cache.add(recordBatch);
                 return cache;
             });
-            int recordSize = recordBatch.size();
-            return size.addAndGet(recordSize + OBJECT_OVERHEAD) >= maxSize || map.size() >= maxStreamCount;
+            return size.addAndGet(recordBatch.occupiedSize()) >= maxSize || map.size() >= maxStreamCount;
         }
 
         public List<StreamRecordBatch> get(long streamId, long startOffset, long endOffset, int maxBytes) {

diff --git a/s3stream/src/main/java/com/automq/stream/s3/model/StreamRecordBatch.java b/s3stream/src/main/java/com/automq/stream/s3/model/StreamRecordBatch.java
@@ -16,7 +16,7 @@
 import io.netty.buffer.ByteBuf;
 
 public class StreamRecordBatch implements Comparable<StreamRecordBatch>, ComparableItem<Long> {
-    public static final int OBJECT_OVERHEAD = 52;
+    private static final int OBJECT_OVERHEAD = 48 /* fields */ + 48 /* ByteBuf payload */ + 48 /* ByteBuf encoded */;
     private final long streamId;
     private final long epoch;
     private final long baseOffset;
@@ -71,6 +71,10 @@ public int size() {
         return payload.readableBytes();
     }
 
+    public int occupiedSize() {
+        return size() + OBJECT_OVERHEAD;
+    }
+
     public void retain() {
         if (encoded != null) {
             encoded.retain();