From 3c37f2f504decedea29e5d8405c15e04b2f603d3 Mon Sep 17 00:00:00 2001 From: caojiajun Date: Fri, 10 Jan 2025 19:37:45 +0800 Subject: [PATCH] feat: fix block cache (#364) --- .../local/storage/file/IBlockReadWrite.java | 7 +++++ .../storage/key/block/KeyBlockReadWrite.java | 29 ++++++++++++------- .../storage/key/persist/KeyFlushExecutor.java | 4 +++ .../string/block/StringBlockReadWrite.java | 7 +++++ 4 files changed, 37 insertions(+), 10 deletions(-) diff --git a/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/file/IBlockReadWrite.java b/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/file/IBlockReadWrite.java index 923608645..28fc0730b 100644 --- a/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/file/IBlockReadWrite.java +++ b/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/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/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/key/block/KeyBlockReadWrite.java b/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/key/block/KeyBlockReadWrite.java index fa22b7c26..f63773630 100644 --- a/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/key/block/KeyBlockReadWrite.java +++ b/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/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 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 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,8 +147,8 @@ 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); @@ -147,7 +156,7 @@ public byte[] getBlock(long fileId, long offset) throws IOException { if (block != null) { return block; } - return fileReadWrite.read(file(fileId), offset, _4k); + return fileReadWrite.read(file(fileId), bucketOffset, _4k); } @Override diff --git a/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/key/persist/KeyFlushExecutor.java b/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/key/persist/KeyFlushExecutor.java index 6d56e730e..833ef1dd0 100644 --- a/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/key/persist/KeyFlushExecutor.java +++ b/camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/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