From 39035f99b7770e16f987932bc644bc893b238642 Mon Sep 17 00:00:00 2001 From: halibobo1205 Date: Wed, 11 Nov 2026 22:17:01 +0800 Subject: [PATCH] perf(db): let SnapshotRoot cache refills share a read lock --- .../org/tron/core/db2/core/SnapshotRoot.java | 32 ++++++--- .../db2/core/SnapshotRootCacheRaceTest.java | 69 +++++++++++++++---- 2 files changed, 75 insertions(+), 26 deletions(-) diff --git a/chainbase/src/main/java/org/tron/core/db2/core/SnapshotRoot.java b/chainbase/src/main/java/org/tron/core/db2/core/SnapshotRoot.java index 3bd7dd1a0cb..6fc0cd417bc 100644 --- a/chainbase/src/main/java/org/tron/core/db2/core/SnapshotRoot.java +++ b/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 { // 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 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 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(); } } } diff --git a/framework/src/test/java/org/tron/core/db2/core/SnapshotRootCacheRaceTest.java b/framework/src/test/java/org/tron/core/db2/core/SnapshotRootCacheRaceTest.java index cd9beeb6914..0ca9ecfdc4a 100644 --- a/framework/src/test/java/org/tron/core/db2/core/SnapshotRootCacheRaceTest.java +++ b/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 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 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 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 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 guava(SnapshotRoot root return (Cache) 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, Future task) - throws InterruptedException { + private static boolean awaitQueuedOrDone(ReentrantReadWriteLock lock, + AtomicReference 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(); }