perf(db): let SnapshotRoot cache refills share a read lock

chainbase/src/main/java/org/tron/core/db2/core/SnapshotRoot.java

@@ -7,6 +7,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.stream.Collectors;
 import lombok.Getter;
 import org.tron.common.cache.CacheManager;
@@ -34,10 +35,11 @@ public class SnapshotRoot extends AbstractSnapshot<byte[], byte[]> {
   // Refill protocol for the shared root cache. Writes to one root must stay serialized
   // (SnapshotManager runs them one at a time); readers call get() concurrently.
   // - get() captures cacheGen before db.get and refills only if cacheGen is unchanged.
-  // - A writer completes the db write, then advances cacheGen under cacheLock, then publishes
-  //   the cache entries.
-  // - The refill rechecks cacheGen and writes the cache while holding the same cacheLock.
-  private final Object cacheLock = new Object();
+  // - A writer completes the db write, then advances cacheGen under the cacheLock write lock,
+  //   then publishes the cache entries.
+  // - The refill rechecks cacheGen and writes the cache under the cacheLock read lock, so
+  //   refills do not block one another.
+  private final ReentrantReadWriteLock cacheLock = new ReentrantReadWriteLock();
   private volatile long cacheGen = 0L;
 
   public SnapshotRoot(DB<byte[], byte[]> db) {
@@ -163,28 +165,36 @@ private boolean cached() {
 
   private void putCache(byte[] key, byte[] value) {
     if (cached()) {
-      synchronized (this.cacheLock) {
-        this.cacheGen++;
-      }
+      advanceCacheGen();
       cache.put(WrappedByteArray.of(key), WrappedByteArray.of(value));
     }
   }
 
   private void putCache(Map<WrappedByteArray, WrappedByteArray> values) {
     if (cached()) {
-      synchronized (this.cacheLock) {
-        this.cacheGen++;
-      }
+      advanceCacheGen();
       values.forEach(cache::put);
     }
   }
 
+  private void advanceCacheGen() {
+    this.cacheLock.writeLock().lock();
+    try {
+      this.cacheGen++;
+    } finally {
+      this.cacheLock.writeLock().unlock();
+    }
+  }
+
   private void refillCache(byte[] key, byte[] value, long observed) {
     if (cached() && this.cacheGen == observed) {
-      synchronized (this.cacheLock) {
+      this.cacheLock.readLock().lock();
+      try {
         if (this.cacheGen == observed) {
           cache.put(WrappedByteArray.of(key), WrappedByteArray.of(value));
         }
+      } finally {
+        this.cacheLock.readLock().unlock();
       }
     }
   }

framework/src/test/java/org/tron/core/db2/core/SnapshotRootCacheRaceTest.java

@@ -1,6 +1,7 @@
 package org.tron.core.db2.core;
 
 import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
@@ -25,6 +26,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Consumer;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -287,15 +289,15 @@ public void staleReadDuringRemovePublishIsSkipped() throws Exception {
   }
 
   /**
-   * A stale reader that passed the lock-free check and then waits on cacheLock must recheck the
-   * generation after acquiring it: a write may advance it while the reader is blocked.
+   * A stale reader that passed the lock-free check and then waits for the read lock must recheck
+   * the generation after acquiring it: a write may advance it while the reader is waiting.
    */
   @Test
   public void staleReadWaitingOnLockIsSkipped() throws Exception {
     FakeFlushDB db = new FakeFlushDB(CACHED_DB);
     db.seed(K, OLD);
     SnapshotRoot root = new SnapshotRoot(db);
-    Object lock = cacheLock(root);
+    ReentrantReadWriteLock lock = cacheLock(root);
 
     CountDownLatch entered = new CountDownLatch(1);
     CountDownLatch release = new CountDownLatch(1);
@@ -308,10 +310,13 @@ public void staleReadWaitingOnLockIsSkipped() throws Exception {
     });
     try {
       assertTrue("reader did not reach the DB gate", entered.await(30, TimeUnit.SECONDS));
-      synchronized (lock) {
-        release.countDown();                 // reader passes the lock-free check, then blocks
-        assertTrue("reader did not block on cacheLock", awaitBlocked(readerThread.get()));
+      lock.writeLock().lock();
+      try {
+        release.countDown();                 // reader passes the lock-free check, then waits
+        assertTrue("reader did not queue on cacheLock", awaitQueued(lock, readerThread.get()));
         root.put(K, NEW);                    // reentrant: advances the generation, caches NEW
+      } finally {
+        lock.writeLock().unlock();
       }
     } finally {
       release.countDown();
@@ -322,9 +327,31 @@ public void staleReadWaitingOnLockIsSkipped() throws Exception {
     WrappedByteArray cached = peek(root, K);
     assertNotNull(cached);
     assertArrayEquals(NEW, cached.getBytes());
+    assertUnlocked(lock);
     assertArrayEquals(NEW, root.get(K));
   }
 
+  /** Refills only share the read lock, so one in flight must not hold up or drop another. */
+  @Test
+  public void concurrentRefillsDoNotBlockEachOther() throws Exception {
+    FakeFlushDB db = new FakeFlushDB(CACHED_DB);
+    db.seed(K, OLD);
+    SnapshotRoot root = new SnapshotRoot(db);
+    ReentrantReadWriteLock lock = cacheLock(root);
+
+    lock.readLock().lock();                  // stands in for another refill in flight
+    try {
+      Future<byte[]> reader = exec.submit(() -> root.get(K));
+      assertArrayEquals(OLD, reader.get(5, TimeUnit.SECONDS));
+      WrappedByteArray cached = peek(root, K);
+      assertNotNull("a concurrent refill must still cache its value", cached);
+      assertArrayEquals(OLD, cached.getBytes());
+    } finally {
+      lock.readLock().unlock();
+    }
+    assertUnlocked(lock);
+  }
+
   /**
    * A reader whose DB read starts while the write is still in flight reads OLD; if its refill lands
    * after the write finishes, it must be skipped. That holds only if the writer completes the DB
@@ -430,6 +457,7 @@ private void assertPublishWaitsForInFlightRefill(Consumer<SnapshotRoot> write)
     FakeFlushDB db = new FakeFlushDB(CACHED_DB);
     db.seed(K, OLD);
     SnapshotRoot root = new SnapshotRoot(db);
+    ReentrantReadWriteLock lock = cacheLock(root);
 
     CountDownLatch refilling = new CountDownLatch(1);
     CountDownLatch resume = new CountDownLatch(1);
@@ -444,7 +472,8 @@ private void assertPublishWaitsForInFlightRefill(Consumer<SnapshotRoot> write)
         writerThread.set(Thread.currentThread());
         write.accept(root);
       });
-      assertTrue("writer neither blocked nor finished", awaitBlockedOrDone(writerThread, writer));
+      assertTrue("writer neither queued nor finished",
+          awaitQueuedOrDone(lock, writerThread, writer));
     } finally {
       resume.countDown();
     }
@@ -454,6 +483,7 @@ private void assertPublishWaitsForInFlightRefill(Consumer<SnapshotRoot> write)
     WrappedByteArray cached = peek(root, K);
     assertNotNull(cached);
     assertArrayEquals(NEW, cached.getBytes());
+    assertUnlocked(lock);
     assertArrayEquals(NEW, root.get(K));
   }
 
@@ -562,6 +592,7 @@ public void missWithoutConcurrentWriteRefills() throws Exception {
     WrappedByteArray cached = peek(root, K);
     assertNotNull("plain miss must refill", cached);
     assertArrayEquals(OLD, cached.getBytes());
+    assertUnlocked(cacheLock(root));
   }
 
   // ---- helpers ----
@@ -584,15 +615,21 @@ private static Cache<WrappedByteArray, WrappedByteArray> guava(SnapshotRoot root
     return (Cache<WrappedByteArray, WrappedByteArray>) guavaField().get(tronCache(root));
   }
 
-  private static Object cacheLock(SnapshotRoot root) throws Exception {
+  private static ReentrantReadWriteLock cacheLock(SnapshotRoot root) throws Exception {
     Field f = SnapshotRoot.class.getDeclaredField("cacheLock");
     f.setAccessible(true);
-    return f.get(root);
+    return (ReentrantReadWriteLock) f.get(root);
+  }
+
+  private static void assertUnlocked(ReentrantReadWriteLock lock) {
+    assertFalse("cacheLock write lock still held", lock.isWriteLocked());
+    assertEquals("cacheLock read lock still held", 0, lock.getReadLockCount());
   }
 
-  private static boolean awaitBlocked(Thread thread) throws InterruptedException {
+  private static boolean awaitQueued(ReentrantReadWriteLock lock, Thread thread)
+      throws InterruptedException {
     long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
-    while (thread.getState() != Thread.State.BLOCKED) {
+    while (!lock.hasQueuedThread(thread)) {
       if (System.nanoTime() > deadline) {
         return false;
       }
@@ -601,12 +638,12 @@ private static boolean awaitBlocked(Thread thread) throws InterruptedException {
     return true;
   }
 
-  private static boolean awaitBlockedOrDone(AtomicReference<Thread> thread, Future<?> task)
-      throws InterruptedException {
+  private static boolean awaitQueuedOrDone(ReentrantReadWriteLock lock,
+      AtomicReference<Thread> thread, Future<?> task) throws InterruptedException {
     long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
     while (!task.isDone()) {
       Thread t = thread.get();
-      if (t != null && t.getState() == Thread.State.BLOCKED) {
+      if (t != null && lock.hasQueuedThread(t)) {
         return true;
       }
       if (System.nanoTime() > deadline) {
@@ -689,7 +726,9 @@ public void put(WrappedByteArray k, WrappedByteArray v) {
   private static void park(CountDownLatch reached, CountDownLatch resume) {
     reached.countDown();
     try {
-      resume.await(30, TimeUnit.SECONDS);
+      if (!resume.await(30, TimeUnit.SECONDS)) {
+        throw new AssertionError("paused thread was never resumed");
+      }
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
     }