Skip to content

Commit 7f33e5e

Browse files
sazzad16sazzad16
authored
Merge b82840a into 3ab6bdc 
2 parents 3ab6bdc + b82840a commit 7f33e5e

File tree

10 files changed

+357
-78
lines changed

10 files changed

+357
-78
lines changed

‎pom.xml‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,27 @@
7575
<version>2.10.1</version>
7676
</dependency>
7777

78+
<!-- Optional dependencies -->
79+
<!-- Client-side caching -->
80+
<dependency>
81+
<groupId>com.google.guava</groupId>
82+
<artifactId>guava</artifactId>
83+
<version>33.0.0-jre</version>
84+
<optional>true</optional>
85+
</dependency>
86+
<dependency>
87+
<groupId>com.github.ben-manes.caffeine</groupId>
88+
<artifactId>caffeine</artifactId>
89+
<version>2.9.3</version>
90+
<optional>true</optional>
91+
</dependency>
92+
<dependency>
93+
<groupId>net.openhft</groupId>
94+
<artifactId>zero-allocation-hashing</artifactId>
95+
<version>0.16</version>
96+
<optional>true</optional>
97+
</dependency>
98+
7899
<!-- UNIX socket connection support -->
79100
<dependency>
80101
<groupId>com.kohlschutter.junixsocket</groupId>
@@ -90,6 +111,7 @@
90111
<version>1.19.0</version>
91112
<scope>test</scope>
92113
</dependency>
114+
93115
<!-- test -->
94116
<dependency>
95117
<groupId>junit</groupId>

‎src/main/java/redis/clients/jedis/ClientSideCache.java‎

Lines changed: 71 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -1,71 +1,104 @@
11
package redis.clients.jedis;
22

33
import java.nio.ByteBuffer;
4-
import java.util.HashMap;
4+
import java.util.HashSet;
55
import java.util.List;
66
import java.util.Map;
7-
8-
import redis.clients.jedis.exceptions.JedisException;
7+
import java.util.Set;
8+
import java.util.concurrent.ConcurrentHashMap;
9+
import java.util.function.Function;
910
import redis.clients.jedis.util.SafeEncoder;
1011

11-
public class ClientSideCache {
12+
/**
13+
* The class to manage the client-side caching. User can provide any of implementation of this class to the client
14+
* object; e.g. {@link redis.clients.jedis.util.CaffeineCSC CaffeineCSC} or
15+
* {@link redis.clients.jedis.util.GuavaCSC GuavaCSC} or a custom implementation of their own.
16+
*/
17+
public abstract class ClientSideCache {
1218

13-
private final Map<ByteBuffer, Object> cache;
19+
protected static final int DEFAULT_MAXIMUM_SIZE = 10_000;
20+
protected static final int DEFAULT_EXPIRE_SECONDS = 100;
1421

15-
public ClientSideCache() {
16-
this.cache = new HashMap<>();
17-
}
22+
private final Map<ByteBuffer, Set<Long>> keyToCommandHashes;
1823

19-
/**
20-
* For testing purpose only.
21-
* @param map
22-
*/
23-
ClientSideCache(Map<ByteBuffer, Object> map) {
24-
this.cache = map;
24+
protected ClientSideCache() {
25+
this.keyToCommandHashes = new ConcurrentHashMap<>();
2526
}
2627

28+
protected abstract void invalidateAllCommandHashes();
29+
30+
protected abstract void invalidateCommandHashes(Iterable<Long> hashes);
31+
32+
protected abstract void put(long hash, Object value);
33+
34+
protected abstract Object get(long hash);
35+
36+
protected abstract long getCommandHash(CommandObject command);
37+
2738
public final void clear() {
28-
cache.clear();
39+
invalidateAllKeysAndCommandHashes();
2940
}
3041

31-
public final void invalidateKeys(List list) {
42+
final void invalidate(List list) {
3243
if (list == null) {
33-
clear();
44+
invalidateAllKeysAndCommandHashes();
3445
return;
3546
}
3647

37-
list.forEach(this::invalidateKey);
48+
list.forEach(this::invalidateKeyAndRespectiveCommandHashes);
3849
}
3950

40-
private void invalidateKey(Object key) {
41-
if (key instanceof byte[]) {
42-
cache.remove(convertKey((byte[]) key));
43-
} else {
44-
throw new JedisException("" + key.getClass().getSimpleName() + " is not supported. Value: " + String.valueOf(key));
45-
}
51+
private void invalidateAllKeysAndCommandHashes() {
52+
invalidateAllCommandHashes();
53+
keyToCommandHashes.clear();
4654
}
4755

48-
protected void setKey(Object key, Object value) {
49-
cache.put(getMapKey(key), value);
50-
}
56+
private void invalidateKeyAndRespectiveCommandHashes(Object key) {
57+
if (!(key instanceof byte[])) {
58+
throw new AssertionError("" + key.getClass().getSimpleName() + " is not supported. Value: " + String.valueOf(key));
59+
}
5160

52-
protected <T> T getValue(Object key) {
53-
return (T) getMapValue(key);
54-
}
61+
final ByteBuffer mapKey = makeKeyForKeyToCommandHashes((byte[]) key);
5562

56-
private Object getMapValue(Object key) {
57-
return cache.get(getMapKey(key));
63+
Set<Long> hashes = keyToCommandHashes.get(mapKey);
64+
if (hashes != null) {
65+
invalidateCommandHashes(hashes);
66+
keyToCommandHashes.remove(mapKey);
67+
}
5868
}
5969

60-
private ByteBuffer getMapKey(Object key) {
61-
if (key instanceof byte[]) {
62-
return convertKey((byte[]) key);
63-
} else {
64-
return convertKey(SafeEncoder.encode(String.valueOf(key)));
70+
final <T> T getValue(Function<CommandObject<T>, T> loader, CommandObject<T> command, String... keys) {
71+
72+
final long hash = getCommandHash(command);
73+
74+
T value = (T) get(hash);
75+
if (value != null) {
76+
return value;
6577
}
78+
79+
value = loader.apply(command);
80+
if (value != null) {
81+
put(hash, value);
82+
for (String key : keys) {
83+
ByteBuffer mapKey = makeKeyForKeyToCommandHashes(key);
84+
if (keyToCommandHashes.containsKey(mapKey)) {
85+
keyToCommandHashes.get(mapKey).add(hash);
86+
} else {
87+
Set<Long> set = new HashSet<>();
88+
set.add(hash);
89+
keyToCommandHashes.put(mapKey, set);
90+
}
91+
}
92+
}
93+
94+
return value;
95+
}
96+
97+
private ByteBuffer makeKeyForKeyToCommandHashes(String key) {
98+
return makeKeyForKeyToCommandHashes(SafeEncoder.encode(key));
6699
}
67100

68-
private static ByteBuffer convertKey(byte[] b) {
101+
private static ByteBuffer makeKeyForKeyToCommandHashes(byte[] b) {
69102
return ByteBuffer.wrap(b);
70103
}
71104
}

‎src/main/java/redis/clients/jedis/Protocol.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -248,7 +248,7 @@ private static void processPush(final RedisInputStream is, ClientSideCache cache
248248
//System.out.println("PUSH: " + SafeEncoder.encodeObject(list));
249249
if (list.size() == 2 && list.get(0) instanceof byte[]
250250
&& Arrays.equals(INVALIDATE_BYTES, (byte[]) list.get(0))) {
251-
cache.invalidateKeys((List) list.get(1));
251+
cache.invalidate((List) list.get(1));
252252
}
253253
}
254254

‎src/main/java/redis/clients/jedis/UnifiedJedis.java‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -295,6 +295,14 @@ public void setBroadcastAndRoundRobinConfig(JedisBroadcastAndRoundRobinConfig co
295295
this.commandObjects.setBroadcastAndRoundRobinConfig(this.broadcastAndRoundRobinConfig);
296296
}
297297

298+
private <T> T executeClientSideCacheCommand(CommandObject<T> command, String... keys) {
299+
if (clientSideCache == null) {
300+
return executeCommand(command);
301+
}
302+
303+
return clientSideCache.getValue((cmd) -> executeCommand(cmd), command, keys);
304+
}
305+
298306
public String ping() {
299307
return checkAndBroadcastCommand(commandObjects.ping());
300308
}
@@ -749,15 +757,7 @@ public String set(String key, String value, SetParams params) {
749757

750758
@Override
751759
public String get(String key) {
752-
if (clientSideCache != null) {
753-
String cachedValue = clientSideCache.getValue(key);
754-
if (cachedValue != null) return cachedValue;
755-
756-
String value = executeCommand(commandObjects.get(key));
757-
if (value != null) clientSideCache.setKey(key, value);
758-
return value;
759-
}
760-
return executeCommand(commandObjects.get(key));
760+
return executeClientSideCacheCommand(commandObjects.get(key), key);
761761
}
762762

763763
@Override

‎src/main/java/redis/clients/jedis/util/CaffeineCSC.java‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
package redis.clients.jedis.util;
2+
3+
import com.github.benmanes.caffeine.cache.Cache;
4+
import com.github.benmanes.caffeine.cache.Caffeine;
5+
import java.util.concurrent.TimeUnit;
6+
import net.openhft.hashing.LongHashFunction;
7+
import redis.clients.jedis.ClientSideCache;
8+
import redis.clients.jedis.CommandObject;
9+
import redis.clients.jedis.args.Rawable;
10+
11+
public class CaffeineCSC extends ClientSideCache {
12+
13+
private static final LongHashFunction DEFAULT_HASH_FUNCTION = LongHashFunction.xx3();
14+
15+
private final Cache<Long, Object> cache;
16+
private final LongHashFunction function;
17+
18+
public CaffeineCSC(Cache<Long, Object> caffeineCache, LongHashFunction hashFunction) {
19+
this.cache = caffeineCache;
20+
this.function = hashFunction;
21+
}
22+
23+
@Override
24+
protected final void invalidateAllCommandHashes() {
25+
cache.invalidateAll();
26+
}
27+
28+
@Override
29+
protected void invalidateCommandHashes(Iterable<Long> hashes) {
30+
cache.invalidateAll(hashes);
31+
}
32+
33+
@Override
34+
protected void put(long hash, Object value) {
35+
cache.put(hash, value);
36+
}
37+
38+
@Override
39+
protected Object get(long hash) {
40+
return cache.getIfPresent(hash);
41+
}
42+
43+
@Override
44+
protected final long getCommandHash(CommandObject command) {
45+
long[] nums = new long[command.getArguments().size() + 1];
46+
int idx = 0;
47+
for (Rawable raw : command.getArguments()) {
48+
nums[idx++] = function.hashBytes(raw.getRaw());
49+
}
50+
nums[idx] = function.hashInt(command.getBuilder().hashCode());
51+
return function.hashLongs(nums);
52+
}
53+
54+
public static Builder builder() {
55+
return new Builder();
56+
}
57+
58+
public static class Builder {
59+
60+
private long maximumSize = DEFAULT_MAXIMUM_SIZE;
61+
private long expireTime = DEFAULT_EXPIRE_SECONDS;
62+
private final TimeUnit expireTimeUnit = TimeUnit.SECONDS;
63+
64+
private LongHashFunction hashFunction = DEFAULT_HASH_FUNCTION;
65+
66+
private Builder() { }
67+
68+
public Builder maximumSize(int size) {
69+
this.maximumSize = size;
70+
return this;
71+
}
72+
73+
public Builder ttl(int seconds) {
74+
this.expireTime = seconds;
75+
return this;
76+
}
77+
78+
public Builder hashFunction(LongHashFunction function) {
79+
this.hashFunction = function;
80+
return this;
81+
}
82+
83+
public CaffeineCSC build() {
84+
Caffeine cb = Caffeine.newBuilder();
85+
86+
cb.maximumSize(maximumSize);
87+
88+
cb.expireAfterWrite(expireTime, expireTimeUnit);
89+
90+
return new CaffeineCSC(cb.build(), hashFunction);
91+
}
92+
}
93+
}

0 commit comments

Comments
 (0)