Add near and remote hit and miss

Author: rbygrave

src/main/java/io/ebean/redis/DuelCache.java

@@ -6,6 +6,7 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.LongAdder;
 
 public class DuelCache implements ServerCache, NearCacheInvalidate {
 
@@ -14,6 +15,11 @@ public class DuelCache implements ServerCache, NearCacheInvalidate {
   private final NearCacheNotify cacheNotify;
   private final String cacheKey;
 
+  private final LongAdder nearMissCount = new LongAdder();
+  private final LongAdder nearHitCount = new LongAdder();
+  private final LongAdder remoteMissCount = new LongAdder();
+  private final LongAdder remoteHitCount = new LongAdder();
+
   public DuelCache(ServerCache near, ServerCache remote, String cacheKey, NearCacheNotify cacheNotify) {
     this.near = near;
     this.remote = remote;
@@ -41,6 +47,9 @@ public Map<Object, Object> getAll(Set<Object> keys) {
 
     Map<Object, Object> resultMap = near.getAll(keys);
 
+    nearHitCount.add(resultMap.size());
+    nearMissCount.add(keys.size() - resultMap.size());
+
     Set<Object> localKeys = resultMap.keySet();
 
     Set<Object> remainingKeys = new HashSet<>();
@@ -52,6 +61,10 @@ public Map<Object, Object> getAll(Set<Object> keys) {
     if (!remainingKeys.isEmpty()) {
       // fetch missing ones from remote cache and merge results
       Map<Object, Object> remoteMap = remote.getAll(remainingKeys);
+
+      remoteHitCount.add(remoteMap.size());
+      remoteMissCount.add(remainingKeys.size() - remoteMap.size());
+
       if (!remoteMap.isEmpty()) {
         near.putAll(remoteMap);
         resultMap.putAll(remoteMap);
@@ -65,10 +78,15 @@ public Map<Object, Object> getAll(Set<Object> keys) {
   public Object get(Object id) {
     Object val = near.get(id);
     if (val != null) {
+      nearHitCount.increment();
       return val;
     }
+    nearMissCount.increment();
     Object remoteVal = remote.get(id);
-    if (remoteVal != null) {
+    if (remoteVal == null) {
+      remoteMissCount.increment();
+    } else {
+      remoteHitCount.increment();
       near.put(id, remoteVal);
     }
     return remoteVal;
@@ -83,7 +101,6 @@ public void putAll(Map<Object, Object> keyValues) {
 
   @Override
   public void put(Object id, Object value) {
-
     near.put(id, value);
     remote.put(id, value);
     cacheNotify.invalidateKey(cacheKey, id);
@@ -126,4 +143,19 @@ public ServerCacheStatistics getStatistics(boolean reset) {
     return null;
   }
 
+  public long getNearMissCount() {
+    return nearMissCount.longValue();
+  }
+
+  public long getNearHitCount() {
+    return nearHitCount.longValue();
+  }
+
+  public long getRemoteMissCount() {
+    return remoteMissCount.longValue();
+  }
+
+  public long getRemoteHitCount() {
+    return remoteHitCount.longValue();
+  }
 }

src/main/java/io/ebean/redis/ModId.java

@@ -0,0 +1,49 @@
+package io.ebean.redis;
+
+import java.security.SecureRandom;
+import java.util.Base64;
+
+/**
+ * Provides a modified base64 encoded UUID and shorter 12 character random unique value.
+ * <p>
+ * <h3>newId()</h3>
+ * <p>
+ * It produces a 22 character string that is a base64 encoded UUID with the +
+ * and / characters replaced with - and _ so as to be URL safe without requiring
+ * encoding.
+ * </p>
+ * <h3>newShortId()</h3>
+ * <p>
+ * It produces a 12 character string that base64 encoded random number (72 bit).
+ * </p>
+ * <p>
+ * Note that this now internally uses java.util.Base64 to encode the values.
+ * </p>
+ */
+public class ModId {
+
+  private static final SecureRandom shortIdSecureRandom = new SecureRandom();
+
+  private static final Base64.Encoder urlEncoder = Base64.getUrlEncoder();
+
+  /**
+   * Return a 12 character string using a 72 bit randomly generated ID encoded
+   * in modified base64.
+   * <p>
+   * A UUID is 128 bits and this is 72 bits so quite a bit smaller but still
+   * very random with one in 4.7 * 10^21 chance of a collision.
+   * </p>
+   */
+  public static String id() {
+
+    // Random 72 bits
+    byte[] randomBytes = new byte[9];
+    shortIdSecureRandom.nextBytes(randomBytes);
+    return encode64(randomBytes);
+  }
+
+  private static String encode64(byte[] bytes) {
+    return urlEncoder.encodeToString(bytes);
+  }
+
+}

src/main/java/io/ebean/redis/RedisCacheFactory.java

@@ -40,9 +40,9 @@
 
 class RedisCacheFactory implements ServerCacheFactory {
 
-  private static final Logger queryLogger = LoggerFactory.getLogger("org.avaje.ebean.cache.QUERY");
+  private static final Logger queryLogger = LoggerFactory.getLogger("io.ebean.cache.QUERY");
 
-  private static final Logger logger = LoggerFactory.getLogger("org.avaje.ebean.cache.CACHE");
+  private static final Logger logger = LoggerFactory.getLogger("io.ebean.cache.CACHE");
 
   private static final Logger tableModLogger = LoggerFactory.getLogger("io.ebean.cache.TABLEMODS");
 
@@ -82,6 +82,8 @@ class RedisCacheFactory implements ServerCacheFactory {
   private final TimedMetric metricMsgOut;
   private final TimedMetric metricMsgIn;
 
+  private final String serverId = ModId.id();
+
   private ServerCacheNotify listener;
 
 
@@ -276,6 +278,7 @@ public void invalidateKeys(String cacheKey, Set<Object> keySet) {
       try (Jedis resource = jedisPool.getResource()) {
         ByteArrayOutputStream ba = new ByteArrayOutputStream(100);
         ObjectOutputStream os = new ObjectOutputStream(ba);
+        os.writeUTF(serverId);
         os.writeInt(MSG_NEARCACHE_KEYS);
         os.writeUTF(cacheKey);
         os.writeInt(keySet.size());
@@ -301,6 +304,7 @@ public void invalidateKey(String cacheKey, Object id) {
       try (Jedis resource = jedisPool.getResource()) {
         ByteArrayOutputStream ba = new ByteArrayOutputStream(100);
         ObjectOutputStream os = new ObjectOutputStream(ba);
+        os.writeUTF(serverId);
         os.writeInt(MSG_NEARCACHE_KEY);
         os.writeUTF(cacheKey);
         os.writeObject(id);
@@ -323,6 +327,7 @@ public void invalidateClear(String cacheKey) {
       try (Jedis resource = jedisPool.getResource()) {
         ByteArrayOutputStream ba = new ByteArrayOutputStream(100);
         ObjectOutputStream os = new ObjectOutputStream(ba);
+        os.writeUTF(serverId);
         os.writeInt(MSG_NEARCACHE_CLEAR);
         os.writeUTF(cacheKey);
         os.flush();
@@ -378,10 +383,15 @@ private void processNearCacheMessage(byte[] message) {
         String cacheKey = null;
         try {
           ObjectInputStream oi = new ObjectInputStream(new ByteArrayInputStream(message));
+          String sourceServerId = oi.readUTF();
           int msgType = oi.readInt();
           cacheKey = oi.readUTF();
+          if (sourceServerId.equals(serverId)) {
+            // ignore this message as we are the server that sent it
+            return;
+          }
           if (logger.isDebugEnabled()) {
-            logger.debug("processNearCacheMessage type:{} cacheKey:{}", msgType, cacheKey);
+            logger.debug("processNearCacheMessage serverId:{} type:{} cacheKey:{}", sourceServerId, msgType, cacheKey);
           }
 
           switch (msgType) {

src/test/java/io/ebean/redis/ClusterTest.java

@@ -7,6 +7,8 @@
 import io.ebean.config.ServerConfig;
 import org.junit.Test;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 public class ClusterTest {
 
 
@@ -23,31 +25,70 @@ private EbeanServer createOther() {
   }
 
   @Test
-  public void test() {
+  public void test() throws InterruptedException {
 
     EbeanServer server = Ebean.getDefaultServer();
 
     EbeanServer other = createOther();
 
     for (int i = 0; i < 10; i++) {
-      EFoo foo = new EFoo("name "+i);
+      EFoo foo = new EFoo("name " + i);
       foo.save();
     }
 
 
     other.getServerCacheManager().clearAll();
 
+    DuelCache dualCache = (DuelCache) other.getServerCacheManager().getBeanCache(EFoo.class);
+
     EFoo foo0 = other.find(EFoo.class, 1);
-    EFoo foo1 = other.find(EFoo.class, 1);
-    EFoo foo2 = other.find(EFoo.class, 1);
+    assertCounts(dualCache, 0, 1, 0, 1);
 
-    other.getServerCacheManager().clearAll();
+    other.find(EFoo.class, 1);
+    assertCounts(dualCache, 1, 1, 0, 1);
+
+    other.find(EFoo.class, 1);
+    assertCounts(dualCache, 2, 1, 0, 1);
+
+    other.find(EFoo.class, 1);
+    assertCounts(dualCache, 3, 1, 0, 1);
+
+    other.find(EFoo.class, 2);
+    assertCounts(dualCache, 3, 2, 0, 2);
+
+
+    //other.getServerCacheManager().clearAll();
 
-    System.out.println("done");
+
+    foo0.setName("name2");
+    foo0.save();
+    allowAsyncMessaging();
 
     EFoo foo3 = other.find(EFoo.class, 1);
+    assertThat(foo3.getName()).isEqualTo("name2");
+    assertCounts(dualCache, 3, 3, 1, 2);
+
+
+    foo0.setName("name3");
+    foo0.save();
+    allowAsyncMessaging();
+
+    foo3 = other.find(EFoo.class, 1);
+    assertThat(foo3.getName()).isEqualTo("name3");
+    assertCounts(dualCache, 3, 4, 2, 2);
 
     System.out.println("done");
+  }
+
+  private void assertCounts(DuelCache dualCache, int nearHits, int nearMiss, int remoteHit, int remoteMiss) {
+
+    assertThat(dualCache.getNearHitCount()).isEqualTo(nearHits);
+    assertThat(dualCache.getNearMissCount()).isEqualTo(nearMiss);
+    assertThat(dualCache.getRemoteHitCount()).isEqualTo(remoteHit);
+    assertThat(dualCache.getRemoteMissCount()).isEqualTo(remoteMiss);
+  }
 
+  private void allowAsyncMessaging() throws InterruptedException {
+    Thread.sleep(200);
   }
 }