-
Notifications
You must be signed in to change notification settings - Fork 143
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
27 changed files
with
844 additions
and
103 deletions.
There are no files selected for viewing
37 changes: 37 additions & 0 deletions
37
...n/java/com/netease/nim/camellia/redis/proxy/upstream/local/storage/codec/StringValue.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| package com.netease.nim.camellia.redis.proxy.upstream.local.storage.codec; | ||
|
|
||
| import com.netease.nim.camellia.codec.Pack; | ||
| import com.netease.nim.camellia.codec.Unpack; | ||
| import com.netease.nim.camellia.redis.proxy.upstream.local.storage.key.KeyInfo; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| /** | ||
| * Created by caojiajun on 2025/1/9 | ||
| */ | ||
| public record StringValue(byte[] key, byte[] value) { | ||
|
|
||
| public static byte[] encode(byte[] key, byte[] value) { | ||
| Pack pack = new Pack(key.length + value.length + 8); | ||
| pack.putVarbin(key); | ||
| pack.putVarbin(value); | ||
| pack.getBuffer().capacity(pack.getBuffer().readableBytes()); | ||
| return pack.getBuffer().array(); | ||
| } | ||
|
|
||
| public static StringValue decode(byte[] data) { | ||
| Unpack unpack = new Unpack(data); | ||
| byte[] key = unpack.popVarbin(); | ||
| byte[] value = unpack.popVarbin(); | ||
| return new StringValue(key, value); | ||
| } | ||
|
|
||
| public static Map<KeyInfo, byte[]> encodeMap(Map<KeyInfo, byte[]> map) { | ||
| Map<KeyInfo, byte[]> result = new HashMap<>(); | ||
| for (Map.Entry<KeyInfo, byte[]> entry : map.entrySet()) { | ||
| result.put(entry.getKey(), StringValue.encode(entry.getKey().getKey(), entry.getValue())); | ||
| } | ||
| return result; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
9 changes: 9 additions & 0 deletions
9
...etease/nim/camellia/redis/proxy/upstream/local/storage/codec/StringValueDecodeResult.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,9 @@ | ||
| package com.netease.nim.camellia.redis.proxy.upstream.local.storage.codec; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Created by caojiajun on 2025/1/9 | ||
| */ | ||
| public record StringValueDecodeResult(List<byte[]> values, int remaining) { | ||
| } |
13 changes: 13 additions & 0 deletions
13
...etease/nim/camellia/redis/proxy/upstream/local/storage/codec/StringValueEncodeResult.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| package com.netease.nim.camellia.redis.proxy.upstream.local.storage.codec; | ||
|
|
||
| import com.netease.nim.camellia.redis.proxy.upstream.local.storage.value.block.BlockInfo; | ||
| import com.netease.nim.camellia.redis.proxy.upstream.local.storage.value.block.ValueLocation; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Created by caojiajun on 2025/1/8 | ||
| */ | ||
| public record StringValueEncodeResult(List<BlockInfo> blockInfos, List<ValueLocation> oldLocations) { | ||
|
|
||
| } |
64 changes: 64 additions & 0 deletions
64
...etease/nim/camellia/redis/proxy/upstream/local/storage/command/CommandOnLocalStorage.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| package com.netease.nim.camellia.redis.proxy.upstream.local.storage.command; | ||
|
|
||
| import com.netease.nim.camellia.redis.proxy.command.Command; | ||
| import com.netease.nim.camellia.redis.proxy.enums.RedisCommand; | ||
| import com.netease.nim.camellia.redis.proxy.reply.Reply; | ||
| import com.netease.nim.camellia.redis.proxy.upstream.local.storage.compact.CompactExecutor; | ||
| import com.netease.nim.camellia.redis.proxy.upstream.local.storage.enums.FlushResult; | ||
| import com.netease.nim.camellia.redis.proxy.upstream.local.storage.key.KeyReadWrite; | ||
| import com.netease.nim.camellia.redis.proxy.upstream.local.storage.value.string.StringReadWrite; | ||
| import com.netease.nim.camellia.redis.proxy.upstream.local.storage.wal.WalGroup; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.concurrent.CompletableFuture; | ||
|
|
||
| /** | ||
| * Created by caojiajun on 2025/1/3 | ||
| */ | ||
| public abstract class CommandOnLocalStorage { | ||
|
|
||
| protected WalGroup walGroup; | ||
|
|
||
| protected CompactExecutor compactExecutor; | ||
|
|
||
| protected KeyReadWrite keyReadWrite; | ||
| protected StringReadWrite stringReadWrite; | ||
|
|
||
| /** | ||
| * redis command of commander | ||
| * @return redis-command | ||
| */ | ||
| public abstract RedisCommand redisCommand(); | ||
|
|
||
| /** | ||
| * check param | ||
| * @param command command | ||
| * @return success or fail | ||
| */ | ||
| protected abstract boolean parse(Command command); | ||
|
|
||
| /** | ||
| * execute command | ||
| * @param slot slot | ||
| * @param command command | ||
| * @return reply | ||
| */ | ||
| protected abstract Reply execute(short slot, Command command) throws Exception; | ||
|
|
||
| /** | ||
| * check and flush after write | ||
| * @param slot slot | ||
| * @throws IOException exception | ||
| */ | ||
| protected void afterWrite(short slot) throws IOException { | ||
| if (keyReadWrite.needFlush(slot) || stringReadWrite.needFlush(slot)) { | ||
| //key flush prepare | ||
| keyReadWrite.flushPrepare(slot); | ||
| //flush string value | ||
| CompletableFuture<FlushResult> future1 = stringReadWrite.flush(slot); | ||
| //flush key | ||
| future1.thenAccept(result -> keyReadWrite.flush(slot)); | ||
| } | ||
| compactExecutor.compact(slot); | ||
| } | ||
| } |
42 changes: 42 additions & 0 deletions
42
...etease/nim/camellia/redis/proxy/upstream/local/storage/command/LocalStorageExecutors.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| package com.netease.nim.camellia.redis.proxy.upstream.local.storage.command; | ||
|
|
||
|
|
||
| import com.netease.nim.camellia.redis.proxy.conf.ProxyDynamicConf; | ||
| import com.netease.nim.camellia.redis.proxy.util.MpscSlotHashExecutor; | ||
| import com.netease.nim.camellia.tools.utils.SysUtils; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * Created by caojiajun on 2025/1/3 | ||
| */ | ||
| public class LocalStorageExecutors { | ||
|
|
||
| private static final Logger logger = LoggerFactory.getLogger(LocalStorageExecutors.class); | ||
|
|
||
| private static volatile LocalStorageExecutors INSTANCE; | ||
|
|
||
| private final MpscSlotHashExecutor commandExecutor; | ||
|
|
||
| private LocalStorageExecutors() { | ||
| int threads = ProxyDynamicConf.getInt("local.storage.command.executor.threads", SysUtils.getCpuNum() * 4); | ||
| int queueSize = ProxyDynamicConf.getInt("local.storage.command.executor.queue.size", 1024*128); | ||
| commandExecutor = new MpscSlotHashExecutor("local-storage-command-executor", threads, queueSize, new MpscSlotHashExecutor.AbortPolicy()); | ||
| logger.info("LocalStorageExecutors init success, threads = {}, queueSize = {}", threads, queueSize); | ||
| } | ||
|
|
||
| public static LocalStorageExecutors getInstance() { | ||
| if (INSTANCE == null) { | ||
| synchronized (LocalStorageExecutors.class) { | ||
| if (INSTANCE == null) { | ||
| INSTANCE = new LocalStorageExecutors(); | ||
| } | ||
| } | ||
| } | ||
| return INSTANCE; | ||
| } | ||
|
|
||
| public MpscSlotHashExecutor getCommandExecutor() { | ||
| return commandExecutor; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.