feat: fix block cache (#364)

netease-im · Jan 10, 2025 · 3c37f2f · 3c37f2f
1 parent d077774
commit 3c37f2f
Show file tree

Hide file tree

Showing 4 changed files with 37 additions and 10 deletions.
diff --git a/...ava/com/netease/nim/camellia/redis/proxy/upstream/local/storage/file/IBlockReadWrite.java b/...ava/com/netease/nim/camellia/redis/proxy/upstream/local/storage/file/IBlockReadWrite.java
@@ -16,6 +16,13 @@ public interface IBlockReadWrite {
      */
     void updateBlockCache(long fileId, long offset, byte[] block);
 
+    /**
+     * clear block cche
+     * @param fileId fileId
+     * @param offset offset
+     */
+    void clearBlockCache(long fileId, long offset);
+
     /**
      * write one or more blocks
      * @param fileId fileId

diff --git a/.../netease/nim/camellia/redis/proxy/upstream/local/storage/key/block/KeyBlockReadWrite.java b/.../netease/nim/camellia/redis/proxy/upstream/local/storage/key/block/KeyBlockReadWrite.java
@@ -17,7 +17,7 @@
 import java.util.Map;
 import java.util.Set;
 
-import static com.netease.nim.camellia.redis.proxy.upstream.local.storage.constants.LocalStorageConstants.*;
+import static com.netease.nim.camellia.redis.proxy.upstream.local.storage.constants.LocalStorageConstants._4k;
 
 /**
  * Created by caojiajun on 2025/1/2
@@ -72,8 +72,9 @@ public KeyInfo get(short slot, Key key) throws IOException {
         int capacity = slotInfo.capacity();
         int bucketSize = capacity / _4k;
         int bucket = KeyHashUtils.hash(key.key()) % bucketSize;
+        long bucketOffset = offset + bucket * _4k;
 
-        String cacheKey = fileId + "|" + offset;
+        String cacheKey = fileId + "|" + bucketOffset;
         byte[] block = readCache.get(cacheKey);
         if (block == null) {
             block = writeCache.get(cacheKey);
@@ -83,7 +84,7 @@ public KeyInfo get(short slot, Key key) throws IOException {
             }
         }
         if (block == null) {
-            block = fileReadWrite.read(file(fileId), offset + bucket * _4k, _4k);
+            block = fileReadWrite.read(file(fileId), bucketOffset, _4k);
             readCache.put(cacheKey, block);
         }
         Map<Key, KeyInfo> map = KeyCodec.decodeBucket(block);
@@ -105,14 +106,15 @@ public KeyInfo getForCompact(short slot, Key key) throws IOException {
         int capacity = slotInfo.capacity();
         int bucketSize = capacity / _4k;
         int bucket = KeyHashUtils.hash(key.key()) % bucketSize;
+        long bucketOffset = offset + bucket * _4k;
 
-        String cacheKey = fileId + "|" + offset;
+        String cacheKey = fileId + "|" + bucketOffset;
         byte[] block = readCache.get(cacheKey);
         if (block == null) {
             block = writeCache.get(cacheKey);
         }
         if (block == null) {
-            block = fileReadWrite.read(file(fileId), offset, bucket * _4k);
+            block = fileReadWrite.read(file(fileId), bucketOffset, _4k);
             writeCache.put(cacheKey, block);
         }
         Map<Key, KeyInfo> map = KeyCodec.decodeBucket(block);
@@ -124,11 +126,18 @@ public KeyInfo getForCompact(short slot, Key key) throws IOException {
     }
 
     @Override
-    public void updateBlockCache(long fileId, long offset, byte[] block) {
+    public void clearBlockCache(long fileId, long bucketOffset) {
+        String cacheKey = fileId + "|" + bucketOffset;
+        readCache.delete(cacheKey);
+        writeCache.delete(cacheKey);
+    }
+
+    @Override
+    public void updateBlockCache(long fileId, long bucketOffset, byte[] block) {
         if (block.length != _4k) {
             return;
         }
-        String cacheKey = fileId + "|" + offset;
+        String cacheKey = fileId + "|" + bucketOffset;
         byte[] cache = readCache.get(cacheKey);
         if (cache != null) {
             readCache.put(cacheKey, block);
@@ -138,16 +147,16 @@ public void updateBlockCache(long fileId, long offset, byte[] block) {
     }
 
     @Override
-    public byte[] getBlock(long fileId, long offset) throws IOException {
-        String cacheKey = fileId + "|" + offset;
+    public byte[] getBlock(long fileId, long bucketOffset) throws IOException {
+        String cacheKey = fileId + "|" + bucketOffset;
         byte[] block = readCache.get(cacheKey);
         if (block == null) {
             block = writeCache.get(cacheKey);
         }
         if (block != null) {
             return block;
         }
-        return fileReadWrite.read(file(fileId), offset, _4k);
+        return fileReadWrite.read(file(fileId), bucketOffset, _4k);
     }
 
     @Override

diff --git a/...netease/nim/camellia/redis/proxy/upstream/local/storage/key/persist/KeyFlushExecutor.java b/...netease/nim/camellia/redis/proxy/upstream/local/storage/key/persist/KeyFlushExecutor.java
@@ -77,6 +77,10 @@ private void clear(SlotInfo slotInfo) throws IOException {
         long fileId = slotInfo.fileId();
         long offset = slotInfo.offset();
         int capacity = slotInfo.capacity();
+        int bucketSize = capacity / _4k;
+        for (int i=0; i<bucketSize; i++) {
+            keyBlockReadWrite.clearBlockCache(fileId, offset + i * _4k);
+        }
         keyBlockReadWrite.writeBlocks(fileId, offset, new byte[capacity]);
     }
 

diff --git a/.../camellia/redis/proxy/upstream/local/storage/value/string/block/StringBlockReadWrite.java b/.../camellia/redis/proxy/upstream/local/storage/value/string/block/StringBlockReadWrite.java
@@ -113,6 +113,13 @@ public void updateBlockCache(long fileId, long offset, byte[] block) {
         }
     }
 
+    @Override
+    public void clearBlockCache(long fileId, long offset) {
+        String key = fileId + "|" + offset;
+        readCache.delete(key);
+        writeCache.delete(key);
+    }
+
     @Override
     public byte[] getBlock(BlockType blockType, long fileId, long offset) throws IOException {
         String cacheKey = fileId + "|" + offset;