From 2430f9271931a628f318f343c5b9c441d10e9d89 Mon Sep 17 00:00:00 2001 From: diegosalvi Date: Tue, 7 Jul 2026 09:35:22 +0200 Subject: [PATCH 1/7] Rework server-side per-key coordination with a non-blocking scheduler Replace the per-key StampedLock (KeyedLockManager) with KeyedScheduler, a per-key event-driven serial scheduler that never blocks a pool thread on a lock across a network round-trip (the root cause of the #188 stalls). - Operations are enqueued per key and drained asynchronously via an iterative, non-reentrant loop; the only lock is a tiny per-slot mutex held for O(1) bookkeeping, never during I/O and never nested, so the design is structurally deadlock-free. - Non-blocking readers-writer scheduling with a FIFO writer barrier: fetches are shared, put/load/invalidate/lock are exclusive, and a queued writer is not starved by a stream of readers. - Aggressive coalescing via submit-time folding under the dominance order INVALIDATE > PUT > LOAD: a dominant write absorbs the redundant broadcasts of contiguous, dominated, not-yet-started predecessors while always preserving their CacheStatus registration; folding never crosses a fetch barrier. - Application locks are owned entirely by the scheduler (held token + owner client id), released on unlock or on disconnect (held, being-acquired, or still-queued), with a liveness check at grant time so a lock is never granted to an already-disconnected client. CacheStatus lock tracking and the LockID type are removed as redundant. - Every completion path is idempotent (reply callbacks may fire more than once) and always resumes the drain (try/finally), so a failed reply cannot strand coalesced callers or stall a key. Adds KeyedSchedulerTest with dedicated coverage for fairness, coalescing, token bypass, idempotency and the lock/disconnect races. Full blazingcache-core suite and the reactor tests (core/jcache/services) pass; apache-rat and spotbugs quality gates are green. --- .../java/blazingcache/server/CacheServer.java | 366 +++----- .../java/blazingcache/server/CacheStatus.java | 82 +- .../blazingcache/server/KeyedLockManager.java | 202 ----- .../blazingcache/server/KeyedScheduler.java | 820 ++++++++++++++++++ .../main/java/blazingcache/server/LockID.java | 67 -- .../server/PendingInvalidationsManager.java | 107 --- .../client/FetchAndInvalidateStormTest.java | 12 +- .../server/KeyedLockManagerLockIdTest.java | 48 - .../server/KeyedSchedulerTest.java | 490 +++++++++++ 9 files changed, 1449 insertions(+), 745 deletions(-) delete mode 100644 blazingcache-core/src/main/java/blazingcache/server/KeyedLockManager.java create mode 100644 blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java delete mode 100644 blazingcache-core/src/main/java/blazingcache/server/LockID.java delete mode 100644 blazingcache-core/src/main/java/blazingcache/server/PendingInvalidationsManager.java delete mode 100644 blazingcache-core/src/test/java/blazingcache/server/KeyedLockManagerLockIdTest.java create mode 100644 blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java index 05c7db0..e654f9c 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java @@ -32,7 +32,6 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -43,6 +42,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; import java.util.logging.Level; import java.util.logging.Logger; import org.apache.zookeeper.ZooKeeper; @@ -60,8 +60,7 @@ public class CacheServer implements AutoCloseable { private final String sharedSecret; private final CacheServerEndpoint acceptor; private final CacheStatus cacheStatus = new CacheStatus(); - private final KeyedLockManager locksManager = new KeyedLockManager(); - private final PendingInvalidationsManager pendingInvalidations = new PendingInvalidationsManager(); + private final KeyedScheduler scheduler = new KeyedScheduler(); private final NettyChannelAcceptor server; private final CacheServerStatusMXBean statusMXBean; private final AtomicLong pendingOperations; @@ -291,153 +290,60 @@ public void close() { } public void putEntry(RawString key, byte[] data, long expiretime, String sourceClientId, String clientProvidedLockId, SimpleCallback onFinish) { - Runnable action = () -> { - final LockID lockID = locksManager.acquireWriteLockForKey(key, sourceClientId, clientProvidedLockId); - if (lockID == null) { - onFinish.onResult(null, new Exception("invalid clientProvidedLockId " + clientProvidedLockId)); + // Registering the source as a holder is the always-applied side effect (kept + // even when the broadcast is coalesced away); the broadcast pushes the new + // value to the other holders. + Runnable registration = () -> cacheStatus.registerKeyForClient(key, sourceClientId, expiretime); + Consumer broadcast = (onComplete) -> { + Set clientsForKey = cacheStatus.getClientsForKey(key); + if (sourceClientId != null) { + clientsForKey.remove(sourceClientId); + } + LOGGER.log(Level.FINEST, "putEntry from {0}, key={1}, clientsForKey:{2}", new Object[]{sourceClientId, key, clientsForKey}); + if (clientsForKey.isEmpty()) { + onComplete.run(); return; } - // finishAndReleaseLock is guarded so the lock is released exactly once, - // whether the release comes from the broadcast completion or from the - // catch block below (StampedLock would throw on a double unlock). - AtomicBoolean finished = new AtomicBoolean(); - SimpleCallback finishAndReleaseLock = (RawString result, Throwable error) -> { - if (finished.compareAndSet(false, true)) { - locksManager.releaseWriteLockForKey(key, sourceClientId, lockID); - onFinish.onResult(result, error); - } - }; - try { - Set clientsForKey = cacheStatus.getClientsForKey(key); - if (sourceClientId != null) { - clientsForKey.remove(sourceClientId); - } - LOGGER.log(Level.FINEST, "putEntry from {0}, key={1}, clientsForKey:{2}", new Object[]{sourceClientId, key, clientsForKey}); - cacheStatus.registerKeyForClient(key, sourceClientId, expiretime); - if (clientsForKey.isEmpty()) { - finishAndReleaseLock.onResult(key, null); - return; + BroadcastRequestStatus propagation = new BroadcastRequestStatus("putEntry " + key + " from " + sourceClientId + " started at " + new java.sql.Timestamp(System.currentTimeMillis()), clientsForKey, (result, error) -> onComplete.run(), null); + networkRequestsStatusMonitor.register(propagation); + + clientsForKey.forEach((clientId) -> { + CacheServerSideConnection connection = acceptor.getActualConnectionFromClient(clientId); + if (connection == null) { + LOGGER.log(Level.SEVERE, "client " + clientId + " not connected, considering key " + key + " invalidated"); + propagation.clientDone(clientId); + } else { + connection.sendPutEntry(sourceClientId, key, data, expiretime, propagation); } - BroadcastRequestStatus propagation = new BroadcastRequestStatus("putEntry " + key + " from " + sourceClientId + " started at " + new java.sql.Timestamp(System.currentTimeMillis()), clientsForKey, finishAndReleaseLock, null); - networkRequestsStatusMonitor.register(propagation); - - clientsForKey.forEach((clientId) -> { - CacheServerSideConnection connection = acceptor.getActualConnectionFromClient(clientId); - if (connection == null) { - LOGGER.log(Level.SEVERE, "client " + clientId + " not connected, considering key " + key + " invalidated"); - propagation.clientDone(clientId); - } else { - connection.sendPutEntry(sourceClientId, key, data, expiretime, propagation); - } - }); - } catch (Throwable t) { - // the body failed before wiring the asynchronous release: free the lock now - LOGGER.log(Level.SEVERE, "error in putEntry for key " + key + ", releasing the lock", t); - finishAndReleaseLock.onResult(null, t); - } + }); }; - executeOnHandler("putEntry " + sourceClientId + "," + key, action); + scheduler.submitExclusive(key, KeyedScheduler.Verb.PUT, clientProvidedLockId, registration, broadcast, key, onFinish); } public void loadEntry(RawString key, long expiretime, String sourceClientId, String clientProvidedLockId, SimpleCallback onFinish) { - Runnable action = () -> { - final LockID lockID = locksManager.acquireWriteLockForKey(key, sourceClientId, clientProvidedLockId); - if (lockID == null) { - onFinish.onResult(null, new Exception("invalid clientProvidedLockId " + clientProvidedLockId)); - return; - } - AtomicBoolean finished = new AtomicBoolean(); - SimpleCallback finishAndReleaseLock = (RawString result, Throwable error) -> { - if (finished.compareAndSet(false, true)) { - locksManager.releaseWriteLockForKey(key, sourceClientId, lockID); - onFinish.onResult(result, error); - } - }; - try { - LOGGER.log(Level.FINEST, "loadEntry from {0}, key={1}", new Object[]{sourceClientId, key}); - cacheStatus.registerKeyForClient(key, sourceClientId, expiretime); - finishAndReleaseLock.onResult(key, null); - } catch (Throwable t) { - LOGGER.log(Level.SEVERE, "error in loadEntry for key " + key + ", releasing the lock", t); - finishAndReleaseLock.onResult(null, t); - } - }; - executeOnHandler("loadEntry " + sourceClientId + "," + key, action); + LOGGER.log(Level.FINEST, "loadEntry from {0}, key={1}", new Object[]{sourceClientId, key}); + // load is local-only: it registers the source as a holder and does not + // broadcast anything to the other clients. + Runnable registration = () -> cacheStatus.registerKeyForClient(key, sourceClientId, expiretime); + scheduler.submitExclusive(key, KeyedScheduler.Verb.LOAD, clientProvidedLockId, registration, null, key, onFinish); } public void invalidateKey(RawString key, String sourceClientId, String clientProvidedLockId, SimpleCallback onFinish) { - // Invalidations carrying an explicit client-provided lock are not coalesced: - // they keep the original per-request semantics tied to that lock. - if (clientProvidedLockId != null) { - invalidateKeyStandalone(key, sourceClientId, clientProvidedLockId, onFinish); - return; - } - Runnable action = () -> { - // Coalesce concurrent invalidations of the same key (issue #188): if an - // invalidation of this key is already in flight, attach to it and let it - // notify us on completion instead of queueing behind the write lock for - // another full broadcast round-trip. - boolean owner = pendingInvalidations.register(key, onFinish); - if (!owner) { - return; - } - final LockID lockID = locksManager.acquireWriteLockForKey(key, sourceClientId); - AtomicBoolean finished = new AtomicBoolean(); - SimpleCallback finishAndReleaseLock = (RawString result, Throwable error) -> { - if (finished.compareAndSet(false, true)) { - cacheStatus.removeKeyForClient(key, sourceClientId); - // Drain the coalesced group BEFORE releasing the lock, so that any - // invalidation arriving during completion (still under the write - // lock, hence still exclusive with fetches) is either captured here - // or starts a fresh broadcast once the lock is released. - List> waiters = pendingInvalidations.complete(key); - locksManager.releaseWriteLockForKey(key, sourceClientId, lockID); - for (SimpleCallback waiter : waiters) { - waiter.onResult(result, error); - } - } - }; - try { - broadcastInvalidation(key, sourceClientId, finishAndReleaseLock); - } catch (Throwable t) { - // release the lock and drain the coalesced group instead of stranding them - LOGGER.log(Level.SEVERE, "error in invalidateKey for key " + key + ", releasing the lock", t); - finishAndReleaseLock.onResult(null, t); - } - }; - executeOnHandler("invalidateKey " + sourceClientId + "," + key, action); - } - - private void invalidateKeyStandalone(RawString key, String sourceClientId, String clientProvidedLockId, SimpleCallback onFinish) { - Runnable action = () -> { - final LockID lockID = locksManager.acquireWriteLockForKey(key, sourceClientId, clientProvidedLockId); - if (lockID == null) { - onFinish.onResult(null, new Exception("invalid clientProvidedLockId " + clientProvidedLockId)); - return; - } - AtomicBoolean finished = new AtomicBoolean(); - SimpleCallback finishAndReleaseLock = (RawString result, Throwable error) -> { - if (finished.compareAndSet(false, true)) { - cacheStatus.removeKeyForClient(key, sourceClientId); - locksManager.releaseWriteLockForKey(key, sourceClientId, lockID); - onFinish.onResult(result, error); - } - }; - try { - broadcastInvalidation(key, sourceClientId, finishAndReleaseLock); - } catch (Throwable t) { - LOGGER.log(Level.SEVERE, "error in invalidateKey for key " + key + ", releasing the lock", t); - finishAndReleaseLock.onResult(null, t); - } - }; - executeOnHandler("invalidateKey " + sourceClientId + "," + key, action); + // Removing the source as a holder is the always-applied side effect (kept even + // when the broadcast is coalesced away); the broadcast tells the other holders + // to drop the entry. Coalescing of concurrent/queued invalidations (issue #188) + // is handled by the scheduler: a plain invalidation carries no client lock and + // is therefore coalescible, while one carrying an explicit lock is not. + Runnable registration = () -> cacheStatus.removeKeyForClient(key, sourceClientId); + Consumer broadcast = (onComplete) -> broadcastInvalidation(key, sourceClientId, (result, error) -> onComplete.run()); + scheduler.submitExclusive(key, KeyedScheduler.Verb.INVALIDATE, clientProvidedLockId, registration, broadcast, key, onFinish); } /** * Snapshots the clients holding the key (excluding the source) and broadcasts * an invalidation to them; {@code onFinish} is invoked once every interested * client has acknowledged (or been considered invalidated). Must be called - * while holding the per-key write lock for {@code key}. + * while the scheduler holds the per-key exclusive slot for {@code key}. */ private void broadcastInvalidation(RawString key, String sourceClientId, SimpleCallback onFinish) { Set clientsForKey = cacheStatus.getClientsForKey(key); @@ -467,109 +373,100 @@ private void broadcastInvalidation(RawString key, String sourceClientId, SimpleC } public void lockKey(RawString key, String sourceClientId, SimpleCallback onFinish) { - Runnable action = () -> { - final LockID lockID = locksManager.acquireWriteLockForKey(key, sourceClientId); - try { - cacheStatus.clientLockedKey(sourceClientId, key, lockID); - onFinish.onResult(lockID.stamp + "", null); - } catch (Throwable t) { - // release the just-acquired lock so the key is not write-locked forever - LOGGER.log(Level.SEVERE, "error in lockKey for key " + key + ", releasing the lock", t); - try { - cacheStatus.clientUnlockedKey(sourceClientId, key, lockID); - locksManager.releaseWriteLockForKey(key, sourceClientId, lockID); - } catch (Throwable release) { - LOGGER.log(Level.SEVERE, "error releasing lock after failed lockKey for key " + key, release); - } - onFinish.onResult(null, t); - } - }; - executeOnHandler("lockKey " + sourceClientId + "," + key, action); + // The scheduler owns the lock lifecycle. It checks, at grant time, that the + // owning client is still connected, so a LOCK request that raced ahead of the + // client's disconnect cleanup is not granted a lock that would never be released. + scheduler.submitLock(key, sourceClientId, + () -> acceptor.getActualConnectionFromClient(sourceClientId) != null, + onFinish); } public void unlockKey(RawString key, String sourceClientId, String lockId, SimpleCallback onFinish) { - Runnable action = () -> { - try { - LockID lockID = new LockID(Long.parseLong(lockId)); - locksManager.releaseWriteLockForKey(key, lockId, lockID); - cacheStatus.clientUnlockedKey(sourceClientId, key, lockID); - onFinish.onResult(lockID.stamp + "", null); - } catch (Throwable t) { - // a malformed lockId or a stale/wrong stamp must not leave the caller hanging - LOGGER.log(Level.SEVERE, "error in unlockKey for key " + key + " lockId " + lockId, t); - onFinish.onResult(null, t); - } - }; - executeOnHandler("unlockKey " + sourceClientId + "," + key, action); + // A malformed lockId must not leave the caller hanging. + Long token = KeyedScheduler.parseProvidedLockId(lockId); + if (token == null) { + onFinish.onResult(null, new Exception("invalid lockId " + lockId)); + return; + } + scheduler.submitUnlock(key, token, onFinish); } public void unregisterEntries(final List keys, String sourceClientId, SimpleCallback onFinish) { LOGGER.log(Level.FINER, "client {0} evicted entries {1}", new Object[]{sourceClientId, keys}); - Runnable action = () -> { - for (RawString key : keys) { - final LockID lockID = locksManager.acquireWriteLockForKey(key, sourceClientId); - try { - cacheStatus.removeKeyForClient(key, sourceClientId); - } finally { - locksManager.releaseWriteLockForKey(key, sourceClientId, lockID); - } - } + if (keys.isEmpty()) { onFinish.onResult(null, null); + return; + } + // Route each eviction through the scheduler as a local (broadcast-less) + // exclusive op so it is serialized with the other work on the same key; reply + // once every key has been processed. + AtomicInteger remaining = new AtomicInteger(keys.size()); + SimpleCallback perKey = (result, error) -> { + if (remaining.decrementAndGet() == 0) { + onFinish.onResult(null, null); + } }; - executeOnHandler("unregisterEntries " + sourceClientId + "," + keys, action); + for (RawString key : keys) { + Runnable registration = () -> cacheStatus.removeKeyForClient(key, sourceClientId); + scheduler.submitExclusive(key, KeyedScheduler.Verb.LOAD, null, registration, null, key, perKey); + } } public void fetchEntry(RawString key, String clientId, String clientProvidedLockId, SimpleCallback onFinish) { - Runnable action = () -> { - // A fetch takes a SHARED (read) lock: concurrent fetches on the same - // key run in parallel, while still being mutually exclusive with - // invalidate/put/load (write lock), preserving the ordering invariant - // that the fetch client-registration must not race an invalidate. - final LockID lockID = locksManager.acquireReadLockForKey(key, clientId, clientProvidedLockId); - if (lockID == null) { - onFinish.onResult(null, new Exception("invalid clientProvidedLockId " + clientProvidedLockId)); - return; - } - AtomicBoolean finished = new AtomicBoolean(); - SimpleCallback finishAndReleaseLock = (Message result, Throwable error) -> { + // A fetch is a SHARED operation: concurrent fetches on the same key run in + // parallel, while still being mutually exclusive with invalidate/put/load + // (exclusive), preserving the ordering invariant that the fetch + // client-registration must not race an invalidate. + // The remote fetch reply callback can fire more than once (send failure then + // reply timeout, see NettyChannel.sendMessageWithAsyncReply); guard so the slot + // release and the client reply (and its pending-operations bookkeeping) happen + // exactly once. + AtomicBoolean finished = new AtomicBoolean(); + Consumer body = (onComplete) -> { + // Register the fetching client (on success) BEFORE releasing the shared + // slot, then reply: a subsequent invalidate can only start once the slot is + // released, so it always sees the registration and removes it. + SimpleCallback finish = (result, error) -> { if (finished.compareAndSet(false, true)) { - locksManager.releaseLockForKey(key, clientId, lockID); + onComplete.run(); onFinish.onResult(result, error); } }; try { - Set clientsForKey = cacheStatus.getClientsForKey(key); - if (clientId != null) { - clientsForKey.remove(clientId); - } - LOGGER.log(Level.FINE, "client {0} fetchEntry {1} ask to {2}", new Object[]{clientId, key, clientsForKey}); - if (clientsForKey.isEmpty()) { - finishAndReleaseLock.onResult(Message.ERROR(clientId, new Exception("no client for key " + key)), null); - return; - } + Set clientsForKey = cacheStatus.getClientsForKey(key); + if (clientId != null) { + clientsForKey.remove(clientId); + } + LOGGER.log(Level.FINE, "client {0} fetchEntry {1} ask to {2}", new Object[]{clientId, key, clientsForKey}); + if (clientsForKey.isEmpty()) { + finish.onResult(Message.ERROR(clientId, new Exception("no client for key " + key)), null); + return; + } - int maxPriority = 0; - List maxPriorityCandidates = new ArrayList<>(); - for (String remoteClientId : clientsForKey) { - CacheServerSideConnection connection = acceptor.getActualConnectionFromClient(remoteClientId); - if (connection != null) { - int fetchPriority = connection.getFetchPriority(); - if (fetchPriority == 0 || fetchPriority < maxPriority) { - continue; - } + int maxPriority = 0; + List maxPriorityCandidates = new ArrayList<>(); + for (String remoteClientId : clientsForKey) { + CacheServerSideConnection connection = acceptor.getActualConnectionFromClient(remoteClientId); + if (connection != null) { + int fetchPriority = connection.getFetchPriority(); + if (fetchPriority == 0 || fetchPriority < maxPriority) { + continue; + } - if (fetchPriority > maxPriority) { - maxPriorityCandidates.clear(); - maxPriority = fetchPriority; + if (fetchPriority > maxPriority) { + maxPriorityCandidates.clear(); + maxPriority = fetchPriority; + } + maxPriorityCandidates.add(connection); } - maxPriorityCandidates.add(connection); } - } - boolean foundOneGoodClientConnected = false; - if (!maxPriorityCandidates.isEmpty()) { - CacheServerSideConnection connection = maxPriorityCandidates.get(ThreadLocalRandom.current().nextInt(maxPriorityCandidates.size())); + if (maxPriorityCandidates.isEmpty()) { + finish.onResult(Message.ERROR(clientId, new Exception("no connected client for key " + key)), null); + return; + } + CacheServerSideConnection connection = maxPriorityCandidates.get(ThreadLocalRandom.current().nextInt(maxPriorityCandidates.size())); String remoteClientId = connection.getClientId(); UnicastRequestStatus unicastRequestStatus = new UnicastRequestStatus(clientId, remoteClientId, "fetch " + key); networkRequestsStatusMonitor.register(unicastRequestStatus); @@ -583,22 +480,15 @@ public void fetchEntry(RawString key, String clientId, String clientProvidedLock long expiretime = (long) result.parameters.get("expiretime"); cacheStatus.registerKeyForClient(key, clientId, expiretime); } - finishAndReleaseLock.onResult(result, error); + finish.onResult(result, error); }); - - foundOneGoodClientConnected = true; - } - - if (!foundOneGoodClientConnected) { - finishAndReleaseLock.onResult(Message.ERROR(clientId, new Exception("no connected client for key " + key)), null); - } } catch (Throwable t) { - // the body failed before wiring the asynchronous release: free the lock now - LOGGER.log(Level.SEVERE, "error in fetchEntry for key " + key + ", releasing the lock", t); - finishAndReleaseLock.onResult(Message.ERROR(clientId, t), null); + LOGGER.log(Level.SEVERE, "error in fetchEntry for key " + key + ", releasing the slot", t); + finish.onResult(Message.ERROR(clientId, t), null); } }; - executeOnHandler("fetchEntry " + clientId + "," + key, action); + scheduler.submitFetch(key, clientProvidedLockId, body, + () -> onFinish.onResult(null, new Exception("invalid clientProvidedLockId " + clientProvidedLockId))); } public void invalidateByPrefix(RawString prefix, String sourceClientId, SimpleCallback onFinish) { @@ -639,18 +529,12 @@ private void executeOnHandler(String name, Runnable runnable) { } void clientDisconnected(String clientId) { - CacheStatus.ClientRemovalResult removalResult = cacheStatus.removeClientListeners(clientId); - int count = removalResult.getListenersCount(); - Map> locks = removalResult.getLocks(); - LOGGER.log(Level.SEVERE, "client " + clientId + " disconnected, removed " + count + " key listeners, locks:" + locks); - if (locks != null) { - locks.forEach((key, locksForKey) -> { - locksForKey.forEach(lock -> { - locksManager.releaseWriteLockForKey(key, clientId, lock); - }); - - }); - } + int count = cacheStatus.removeClientListeners(clientId); + LOGGER.log(Level.SEVERE, "client " + clientId + " disconnected, removed " + count + " key listeners"); + // Release every application lock the client held or was queued for. This is + // driven entirely by the scheduler (which owns the lock ownership), so a lock + // being acquired or still queued when the client dropped is not left dangling. + scheduler.releaseLocksForClient(clientId); } public long getCurrentTimestamp() { @@ -681,12 +565,12 @@ public String getServerId() { return this.serverId; } - public KeyedLockManager getLocksManager() { - return locksManager; + public KeyedScheduler getLocksManager() { + return scheduler; } public int getNumberOfLockedKeys() { - return this.locksManager.getNumberOfLockedKeys(); + return this.scheduler.getNumberOfLockedKeys(); } public long getSlowClientTimeout() { diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java b/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java index 01f50a4..9564652 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java @@ -28,7 +28,6 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.logging.Level; import java.util.logging.Logger; @@ -47,8 +46,6 @@ public class CacheStatus { private final Map> keysForClient = new HashMap<>(); private final Map entryExpireTime = new HashMap<>(); private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true); - private final Map>> remoteLocks = new HashMap<>(); - private final ReentrantLock remoteLocksLock = new ReentrantLock(true); @Override public String toString() { @@ -190,27 +187,13 @@ public void removeKeyByPrefixForClient(RawString prefix, String client) { } } - public static final class ClientRemovalResult { - - private final int listenersCount; - private final Map> locks; - - public ClientRemovalResult(int listenersCount, Map> locks) { - this.listenersCount = listenersCount; - this.locks = locks; - } - - public int getListenersCount() { - return listenersCount; - } - - public Map> getLocks() { - return locks; - } - - } - - ClientRemovalResult removeClientListeners(String clientId) { + /** + * Removes all the key listeners of a disconnected client. + * + * @return the number of listeners removed. Application locks held by the client are + * released separately by the {@link KeyedScheduler}, which owns the lock lifecycle. + */ + int removeClientListeners(String clientId) { AtomicInteger count = new AtomicInteger(); lock.writeLock().lock(); try { @@ -232,14 +215,7 @@ ClientRemovalResult removeClientListeners(String clientId) { } finally { lock.writeLock().unlock(); } - Map> locksForClient; - remoteLocksLock.lock(); - try { - locksForClient = remoteLocks.remove(clientId); - } finally { - remoteLocksLock.unlock(); - } - return new ClientRemovalResult(count.get(), locksForClient); + return count.get(); } Set getAllClientsWithListener() { @@ -277,46 +253,4 @@ void touchKeyFromClient(RawString key, String clientId, long expiretime) { lock.writeLock().unlock(); } } - - void clientLockedKey(String sourceClientId, RawString key, LockID lockID) { - remoteLocksLock.lock(); - try { - Map> locksForClient = remoteLocks.get(sourceClientId); - if (locksForClient == null) { - locksForClient = new HashMap<>(); - remoteLocks.put(sourceClientId, locksForClient); - } - List listForKey = locksForClient.get(key); - if (listForKey == null) { - listForKey = new ArrayList<>(); - locksForClient.put(key, listForKey); - } - listForKey.add(lockID); - } finally { - remoteLocksLock.unlock(); - } - } - - void clientUnlockedKey(String sourceClientId, RawString key, LockID lockID) { - remoteLocksLock.lock(); - try { - Map> locksForClient = remoteLocks.get(sourceClientId); - if (locksForClient == null) { - return; - } - List listForKey = locksForClient.get(key); - if (listForKey == null) { - return; - } - listForKey.remove(lockID); - if (listForKey.isEmpty()) { - locksForClient.remove(key); - if (locksForClient.isEmpty()) { - remoteLocks.remove(sourceClientId); - } - } - } finally { - remoteLocksLock.unlock(); - } - } } diff --git a/blazingcache-core/src/main/java/blazingcache/server/KeyedLockManager.java b/blazingcache-core/src/main/java/blazingcache/server/KeyedLockManager.java deleted file mode 100644 index 0e49ee0..0000000 --- a/blazingcache-core/src/main/java/blazingcache/server/KeyedLockManager.java +++ /dev/null @@ -1,202 +0,0 @@ -/* - Licensed to Diennea S.r.l. under one - or more contributor license agreements. See the NOTICE file - distributed with this work for additional information - regarding copyright ownership. Diennea S.r.l. licenses this file - to you under the Apache License, Version 2.0 (the - "License"); you may not use this file except in compliance - with the License. You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, - software distributed under the License is distributed on an - "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - KIND, either express or implied. See the License for the - specific language governing permissions and limitations - under the License. - - */ -package blazingcache.server; - -import blazingcache.utils.RawString; -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.locks.ReentrantLock; -import java.util.concurrent.locks.StampedLock; -import java.util.logging.Level; -import java.util.logging.Logger; - -/** - * Handle locks by key - * - * @author enrico.olivelli - */ -public class KeyedLockManager { - - private static final Logger LOGGER = Logger.getLogger(KeyedLockManager.class.getName()); - - private StampedLock makeLock() { - return new StampedLock(); - } - private final ReentrantLock generalLock = new ReentrantLock(true); - private final Map liveLocks = new HashMap<>(); - private final Map locksCounter = new HashMap<>(); - - /** - * Debug operation to see actual locked keys - * - * @return - */ - public Map getLockedKeys() { - HashMap result = new HashMap<>(); - generalLock.lock(); - try { - locksCounter.forEach((k, v) -> { - result.put(k, v.get()); - }); - } finally { - generalLock.unlock(); - } - return result; - } - - /** - * - * @return the number of currently locked keys - */ - public int getNumberOfLockedKeys() { - this.generalLock.lock(); - try { - return this.locksCounter.size(); - } finally { - this.generalLock.unlock(); - } - } - - private StampedLock makeLockForKey(RawString key) { - StampedLock lock; - generalLock.lock(); - try { - lock = liveLocks.get(key); - if (lock == null) { - lock = makeLock(); - liveLocks.put(key, lock); - locksCounter.put(key, new AtomicInteger(1)); - } else { - locksCounter.get(key).incrementAndGet(); - } - } finally { - generalLock.unlock(); - } - return lock; - } - - private StampedLock getLockForKey(RawString key) { - StampedLock lock; - generalLock.lock(); - try { - lock = liveLocks.get(key); - } finally { - generalLock.unlock(); - } - return lock; - } - - private StampedLock returnLockForKey(RawString key) throws IllegalStateException { - StampedLock lock; - generalLock.lock(); - try { - lock = liveLocks.get(key); - if (lock == null) { - LOGGER.log(Level.SEVERE, "no lock object exists for key {0}", key); - throw new IllegalStateException("no lock object exists for key " + key); - } - int actualCount = locksCounter.get(key).decrementAndGet(); - if (actualCount == 0) { - liveLocks.remove(key); - locksCounter.remove(key); - } - } finally { - generalLock.unlock(); - } - return lock; - } - - LockID acquireWriteLockForKey(RawString key, String clientId, String clientProvidedLockId) { - if (clientProvidedLockId != null) { - Long stamp = parseLockId(clientProvidedLockId); - return stamp == null ? null : useClientProvidedLockForKey(key, stamp); - } else { - return acquireWriteLockForKey(key, clientId); - } - } - - /** - * Parses a client-provided lock id, returning {@code null} (treated as an - * invalid lock) instead of throwing on a malformed value, so a bad id is - * reported to the client rather than left hanging until timeout. - */ - private static Long parseLockId(String clientProvidedLockId) { - try { - return Long.parseLong(clientProvidedLockId); - } catch (NumberFormatException e) { - LOGGER.log(Level.SEVERE, "invalid clientProvidedLockId {0}", clientProvidedLockId); - return null; - } - } - - LockID acquireWriteLockForKey(RawString key, String clientId) { - StampedLock lock = makeLockForKey(key); - LockID result = new LockID(lock.writeLock()); - return result; - } - - /** - * Acquires a shared (read) lock for a key. Read locks are mutually exclusive - * with write locks (invalidate/put/load) but NOT with each other, so multiple - * concurrent fetches on the same key no longer serialize. See issue #188. - */ - LockID acquireReadLockForKey(RawString key, String clientId, String clientProvidedLockId) { - if (clientProvidedLockId != null) { - Long stamp = parseLockId(clientProvidedLockId); - return stamp == null ? null : useClientProvidedLockForKey(key, stamp); - } else { - return acquireReadLockForKey(key, clientId); - } - } - - LockID acquireReadLockForKey(RawString key, String clientId) { - StampedLock lock = makeLockForKey(key); - LockID result = new LockID(lock.readLock()); - return result; - } - - void releaseWriteLockForKey(RawString key, String clientId, LockID lockStamp) { - releaseLockForKey(key, clientId, lockStamp); - } - - /** - * Releases a lock (read or write) previously acquired for the key. - * {@link StampedLock#unlock(long)} releases the correct mode according to the - * stamp, so the same method works for both read and write locks. - */ - void releaseLockForKey(RawString key, String clientId, LockID lockStamp) { - if (lockStamp == LockID.VALIDATED_CLIENT_PROVIDED_LOCK) { - return; - } - StampedLock lock = returnLockForKey(key); - lock.unlock(lockStamp.stamp); - } - - LockID useClientProvidedLockForKey(RawString key, long stamp) { - StampedLock lock = getLockForKey(key); - if (lock.validate(stamp)) { - return LockID.VALIDATED_CLIENT_PROVIDED_LOCK; - } else { - return null; - } - } - -} diff --git a/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java b/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java new file mode 100644 index 0000000..29dfa20 --- /dev/null +++ b/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java @@ -0,0 +1,820 @@ +/* + Licensed to Diennea S.r.l. under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. Diennea S.r.l. licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + + */ +package blazingcache.server; + +import blazingcache.utils.RawString; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.BooleanSupplier; +import java.util.function.Consumer; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Per-key event-driven serial scheduler that coordinates the cache operations + * (fetch/put/load/invalidate/lock/unlock) on the {@link CacheServer} without ever + * blocking a thread on a lock across a network round-trip. + *

+ * Each key owns a {@link KeySlot} holding a FIFO queue drained asynchronously: + * submitting an operation enqueues it and returns immediately; when the running + * operation completes (via its completion callback, typically fired by the network + * layer) the next operation is started. The only lock ever held is the tiny per-slot + * {@link ReentrantLock}, taken exclusively for O(1) queue/counter bookkeeping and + * never while performing I/O and never nested with any other lock, so the design is + * structurally deadlock-free. + *

+ * Scheduling is a non-blocking readers-writer discipline: fetches are + * {@link Mode#SHARED} (run concurrently), while put/load/invalidate/lock are + * {@link Mode#EXCLUSIVE}. FIFO ordering with a writer barrier prevents writer + * starvation: readers that arrive while a writer is already queued are ordered behind + * it. The drain loop is iterative and non-reentrant, so a run of synchronously + * completing operations cannot overflow the stack. + *

+ * The scheduler also performs aggressive coalescing of work on the same key. + * Each write op has two separable effects: (a) a network broadcast to the interested + * clients and (b) a {@code CacheStatus} registration mutation. Coalescing may drop + * only the broadcast (a); the registration (b) is always preserved. Under the + * dominance order {@code INVALIDATE > PUT > LOAD} a newly submitted write folds the + * contiguous, dominated, not-yet-started predecessors on the tail of the queue into + * itself (stopping at a fetch or a non-dominated op): their registrations are replayed + * in order right before the dominant op's single broadcast, and their callers are + * notified when that broadcast completes. Additionally, plain invalidations arriving + * while an invalidation of the same key is in flight attach to it (idempotency) + * instead of queueing another full broadcast. + * + * @author diego.salvi + */ +public final class KeyedScheduler { + + private static final Logger LOGGER = Logger.getLogger(KeyedScheduler.class.getName()); + + enum Mode { + SHARED, EXCLUSIVE + } + + enum Verb { + FETCH, PUT, LOAD, INVALIDATE, LOCK + } + + private final ConcurrentHashMap slots = new ConcurrentHashMap<>(); + private final AtomicLong tokenGenerator = new AtomicLong(); + + /** + * Parses a client-provided lock id, returning {@code null} (treated as an invalid + * lock) instead of throwing on a malformed value, so a bad id is reported to the + * client rather than left hanging until timeout (issue #188). + */ + static Long parseProvidedLockId(String clientProvidedLockId) { + try { + return Long.parseLong(clientProvidedLockId); + } catch (NumberFormatException e) { + LOGGER.log(Level.SEVERE, "invalid clientProvidedLockId {0}", clientProvidedLockId); + return null; + } + } + + // ------------------------------------------------------------------ + // Public API used by CacheServer + // ------------------------------------------------------------------ + + /** + * Submits a fetch (shared) operation. {@code body} receives a completion callback + * it must invoke exactly once when the asynchronous fetch has been fully handled. + * If a client-provided lock id is given and valid, the op bypasses serialization + * (the owner already holds exclusivity); if it is malformed or does not match the + * held lock, {@code onInvalidLock} is invoked instead of {@code body}. + */ + public void submitFetch(RawString key, String providedLockId, + Consumer body, Runnable onInvalidLock) { + Op op = new Op(Verb.FETCH, Mode.SHARED); + op.sharedBody = body; + submit(key, op, providedLockId, onInvalidLock); + } + + /** + * Submits an exclusive write operation (put/load/invalidate). + * + * @param registration synchronous CacheStatus mutation applied, in queue order, + * right before the broadcast (may be {@code null}); always executed even when the + * broadcast is coalesced away. + * @param broadcast asynchronous network work; must invoke the supplied callback + * once completed. {@code null} means there is no broadcast (e.g. load). + * @param onFinish invoked with {@code (resultKey, null)} on success or + * {@code (null, error)} on failure, once this op (and every op coalesced into it) + * is complete. + */ + public void submitExclusive(RawString key, Verb verb, String providedLockId, + Runnable registration, Consumer broadcast, + RawString resultKey, SimpleCallback onFinish) { + Op op = new Op(verb, Mode.EXCLUSIVE); + op.registration = registration; + op.broadcast = broadcast; + op.resultKey = resultKey; + op.onFinish = onFinish; + op.coalescible = providedLockId == null + && (verb == Verb.PUT || verb == Verb.LOAD || verb == Verb.INVALIDATE); + submit(key, op, providedLockId, + () -> onFinish.onResult(null, new Exception("invalid clientProvidedLockId " + providedLockId))); + } + + /** + * Acquires an application lock on the key. When the lock op reaches the head of the + * queue a fresh monotonic token is published atomically as the held token; the slot + * then stays "held" (no further queued op is drained) until {@link #submitUnlock} + * (or {@link #releaseLocksForClient} on disconnect) releases it. Operations carrying + * the matching token bypass the queue in the meantime. + *

+ * {@code ownerStillConnected} is checked at grant time: if the owning client has + * disconnected by then (its LOCK request having raced ahead of the disconnect + * cleanup), the lock is immediately released instead of being granted to a gone + * client, closing the acquire-vs-disconnect window. + */ + public void submitLock(RawString key, String clientId, + BooleanSupplier ownerStillConnected, SimpleCallback onFinish) { + long token = tokenGenerator.incrementAndGet(); + Op op = new Op(Verb.LOCK, Mode.EXCLUSIVE); + op.sourceClientId = clientId; + op.lockToken = token; + op.ownerStillConnected = ownerStillConnected; + op.lockOnFinish = onFinish; + while (true) { + KeySlot slot = slots.computeIfAbsent(key, KeySlot::new); + slot.lock.lock(); + try { + if (slot.removed) { + continue; + } + slot.queue.add(op); + } finally { + slot.lock.unlock(); + } + drain(slot); + return; + } + } + + /** + * Releases an application lock previously acquired with the given token and resumes + * the drain of the queued operations. Replies an error if no lock with that token is + * currently held on the key. + */ + public void submitUnlock(RawString key, long token, SimpleCallback onFinish) { + KeySlot slot = slots.get(key); + boolean released = false; + if (slot != null) { + slot.lock.lock(); + try { + // guard against token 0, the "no lock held" sentinel, so unlocking an + // unlocked key does not produce a false success + if (token != 0 && slot.heldToken == token) { + slot.heldToken = 0; + slot.lockOwnerClientId = null; + released = true; + } + } finally { + slot.lock.unlock(); + } + } + if (released) { + // The lock is already released; make sure the queued work behind it is + // resumed (drain) even if the reply throws, otherwise the key would stall + // until an unrelated later submit. + try { + onFinish.onResult(Long.toString(token), null); + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error replying to unlock for key " + key, t); + } finally { + drain(slot); + } + } else { + onFinish.onResult(null, new Exception("no lock held for key " + key + " with id " + token)); + } + } + + /** + * Releases every application lock owned by a disconnected client and cancels its + * still-queued lock requests, then resumes the drain of the affected keys. This is + * the single, scheduler-driven release path used on disconnect: because the lock + * ownership ({@code heldToken} + {@code lockOwnerClientId}) is published atomically + * at dequeue, there is no window in which a held or being-acquired lock is invisible + * here, and queued lock requests that never ran are cancelled rather than later + * granted to the gone client. + */ + public void releaseLocksForClient(String clientId) { + slots.forEach((key, slot) -> { + List> cancelledLocks = null; + boolean changed = false; + slot.lock.lock(); + try { + if (slot.heldToken != 0 && clientId.equals(slot.lockOwnerClientId)) { + slot.heldToken = 0; + slot.lockOwnerClientId = null; + changed = true; + } + for (Iterator it = slot.queue.iterator(); it.hasNext();) { + Op queued = it.next(); + if (queued.verb == Verb.LOCK && clientId.equals(queued.sourceClientId)) { + it.remove(); + if (cancelledLocks == null) { + cancelledLocks = new ArrayList<>(); + } + cancelledLocks.add(queued.lockOnFinish); + changed = true; + } + } + } finally { + slot.lock.unlock(); + } + if (cancelledLocks != null) { + for (SimpleCallback cancelled : cancelledLocks) { + try { + cancelled.onResult(null, new Exception("client " + clientId + + " disconnected while waiting for the lock on " + key)); + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error cancelling queued lock for key " + key, t); + } + } + } + if (changed) { + drain(slot); + } + }); + } + + // ------------------------------------------------------------------ + // Introspection (compatibility with legacy KeyedLockManager / JMX) + // ------------------------------------------------------------------ + + /** + * Debug view of the currently "locked" keys and the number of outstanding + * operations referencing each of them (in-flight + queued + a held application + * lock). Operations coalesced into an in-flight or dominant one are not counted, + * mirroring the legacy behaviour where only the broadcast owner held the lock. + */ + public Map getLockedKeys() { + HashMap result = new HashMap<>(); + slots.forEach((key, slot) -> { + slot.lock.lock(); + try { + int count = slot.activeReaders + (slot.activeWriter ? 1 : 0) + + slot.queue.size() + (slot.heldToken != 0 ? 1 : 0) + slot.activeBypass; + if (count > 0) { + result.put(key, count); + } + } finally { + slot.lock.unlock(); + } + }); + return result; + } + + /** + * @return the number of keys currently tracked by the scheduler (at rest, the keys + * holding an application lock). + */ + public int getNumberOfLockedKeys() { + return slots.size(); + } + + // ------------------------------------------------------------------ + // Internal machinery + // ------------------------------------------------------------------ + + private void submit(RawString key, Op op, String providedLockId, Runnable onInvalidLock) { + Long token = null; + if (providedLockId != null) { + token = parseProvidedLockId(providedLockId); + if (token == null) { + onInvalidLock.run(); + return; + } + } + while (true) { + KeySlot slot = slots.computeIfAbsent(key, KeySlot::new); + boolean bypass = false; + boolean invalid = false; + boolean enqueued = false; + slot.lock.lock(); + try { + if (slot.removed) { + continue; + } + if (token != null) { + if (slot.heldToken != 0 && slot.heldToken == token) { + bypass = true; + // count the in-flight bypass op so a concurrent unlock/disconnect + // does not let queued exclusive ops start before it completes + slot.activeBypass++; + } else { + invalid = true; + } + } else if (op.coalescible && op.verb == Verb.INVALIDATE + && slot.inflightInvalidate != null && slot.inflightInvalidate.acceptingCoalesce) { + // idempotent invalidation already in flight: attach and let it notify us + slot.inflightInvalidate.coalescedWaiters().add(op.onFinish); + } else { + if (op.coalescible) { + foldDominatedTail(slot, op); + } + slot.queue.add(op); + enqueued = true; + } + } finally { + slot.lock.unlock(); + } + if (invalid) { + onInvalidLock.run(); + maybeRemove(slot); + } else if (bypass) { + runBypass(slot, op); + } else if (enqueued) { + drain(slot); + } + // (attached) -> nothing: our callback will be invoked by the in-flight invalidation + return; + } + } + + /** + * Folds the contiguous, dominated, not-yet-started predecessors on the tail of the + * queue into {@code x}: they are removed from the queue, their registrations are + * collected (in chronological order) to be replayed before {@code x}'s broadcast, + * and their callers are collected to be notified when {@code x} completes. Scanning + * stops at the first fetch (shared) or non-dominated op. + * Must be called while holding {@code slot.lock}. + */ + private void foldDominatedTail(KeySlot slot, Op x) { + // Fast path (by far the most common): nothing on the tail can be folded. + Op tail = slot.queue.peekLast(); + if (tail == null || tail.mode != Mode.EXCLUSIVE || tail.verb == Verb.LOCK + || !tail.coalescible || !dominates(x.verb, tail.verb)) { + return; + } + ArrayDeque absorbedNewestFirst = new ArrayDeque<>(); + Iterator it = slot.queue.descendingIterator(); + while (it.hasNext()) { + Op t = it.next(); + if (t.mode != Mode.EXCLUSIVE || t.verb == Verb.LOCK || !t.coalescible || !dominates(x.verb, t.verb)) { + break; + } + absorbedNewestFirst.add(t); + } + for (int i = 0; i < absorbedNewestFirst.size(); i++) { + slot.queue.pollLast(); + } + // replay in chronological order (oldest first) + Iterator chronological = absorbedNewestFirst.descendingIterator(); + while (chronological.hasNext()) { + Op t = chronological.next(); + if (t.foldedRegistrations != null) { + x.foldedRegistrations().addAll(t.foldedRegistrations); + } + if (t.registration != null) { + x.foldedRegistrations().add(t.registration); + } + if (t.coalescedWaiters != null) { + x.coalescedWaiters().addAll(t.coalescedWaiters); + } + x.coalescedWaiters().add(t.onFinish); + } + } + + private static boolean dominates(Verb x, Verb t) { + switch (x) { + case INVALIDATE: + return t == Verb.INVALIDATE || t == Verb.PUT || t == Verb.LOAD; + case PUT: + return t == Verb.PUT || t == Verb.LOAD; + default: + return false; + } + } + + /** + * Iterative, non-reentrant drain: selects and starts the runnable operations of a + * slot. Only one thread drains a given slot at a time; completions that fire while + * a drain is in progress (including synchronous ones triggered from within a + * started body) simply request another pass instead of recursing. + */ + private void drain(KeySlot slot) { + slot.lock.lock(); + try { + if (slot.draining) { + slot.drainAgain = true; + return; + } + slot.draining = true; + } finally { + slot.lock.unlock(); + } + while (true) { + List toStart = new ArrayList<>(); + slot.lock.lock(); + try { + slot.drainAgain = false; + selectRunnable(slot, toStart); + } finally { + slot.lock.unlock(); + } + for (Op op : toStart) { + runStarted(slot, op); + } + slot.lock.lock(); + try { + if (!slot.drainAgain) { + slot.draining = false; + break; + } + } finally { + slot.lock.unlock(); + } + } + maybeRemove(slot); + } + + /** + * Selects the next operations to start, updating the slot counters. Must be called + * while holding {@code slot.lock}. + */ + private void selectRunnable(KeySlot slot, List toStart) { + // Do not start queued work while the key is application-locked (heldToken) or + // while lock-bypass operations of the current owner are still in flight + // (activeBypass): the latter keeps the readers-writer barrier sound across a + // concurrent unlock/disconnect. + if (slot.heldToken != 0 || slot.activeBypass > 0) { + return; + } + while (true) { + Op head = slot.queue.peek(); + if (head == null) { + return; + } + if (head.mode == Mode.EXCLUSIVE) { + if (slot.activeReaders > 0 || slot.activeWriter) { + return; + } + slot.queue.poll(); + if (head.verb == Verb.LOCK) { + // Publish the lock ownership atomically at dequeue: this is the + // single point that makes the lock visible, so a concurrent + // disconnect can always release it and a token-carrying op cannot + // race ahead of the acquisition. No activeWriter is taken, so the + // key is not double-counted while runLock records it. + slot.heldToken = head.lockToken; + slot.lockOwnerClientId = head.sourceClientId; + } else { + slot.activeWriter = true; + if (head.verb == Verb.INVALIDATE && head.coalescible) { + slot.inflightInvalidate = head; + head.acceptingCoalesce = true; + } + } + toStart.add(head); + return; + } else { + if (slot.activeWriter) { + return; + } + slot.queue.poll(); + slot.activeReaders++; + toStart.add(head); + } + } + } + + private void runStarted(KeySlot slot, Op op) { + if (op.mode == Mode.SHARED) { + try { + op.sharedBody.accept(() -> completeShared(slot, op)); + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error running fetch for key " + op.resultKey, t); + completeShared(slot, op); + } + return; + } + if (op.verb == Verb.LOCK) { + runLock(slot, op); + } else { + runExclusive(slot, op); + } + } + + private void runExclusive(KeySlot slot, Op op) { + try { + if (op.foldedRegistrations != null) { + for (Runnable registration : op.foldedRegistrations) { + registration.run(); + } + } + if (op.registration != null) { + op.registration.run(); + } + Consumer broadcast = op.broadcast; + if (broadcast == null) { + completeExclusive(slot, op, op.resultKey, null); + } else { + broadcast.accept(() -> completeExclusive(slot, op, op.resultKey, null)); + } + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error running " + op.verb + " for key " + op.resultKey + ", releasing the slot", t); + completeExclusive(slot, op, null, t); + } + } + + private void runLock(KeySlot slot, Op op) { + // heldToken + lockOwnerClientId were published at dequeue (selectRunnable). Only + // keep the lock if we still hold it AND the owning client is still connected: a + // concurrent disconnect may already have released it, or the LOCK request may + // have raced ahead of the disconnect cleanup (which runs after the connection is + // removed), in which case the owner is already gone. + // evaluate liveness outside the slot lock (it calls into the acceptor); the tiny + // window between this check and the grant is backstopped by releaseLocksForClient, + // which runs after the connection is removed and finds the published heldToken. + boolean ownerConnected = op.ownerStillConnected == null || op.ownerStillConnected.getAsBoolean(); + boolean granted = false; + slot.lock.lock(); + try { + if (slot.heldToken == op.lockToken) { + if (ownerConnected) { + granted = true; + } else { + // release the lock we just published: the owner disconnected + slot.heldToken = 0; + slot.lockOwnerClientId = null; + } + } + } finally { + slot.lock.unlock(); + } + boolean replied = false; + try { + if (granted) { + op.lockOnFinish.onResult(Long.toString(op.lockToken), null); + } else { + op.lockOnFinish.onResult(null, new Exception("lock acquisition aborted for key " + slot.key + + " (owner disconnected)")); + } + replied = true; + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error replying to lockKey for key " + slot.key, t); + } finally { + // If the grant reply failed to reach the client, release the just-acquired + // lock: the client does not know it holds it, so leaving heldToken set would + // stall the key forever (the client stays connected, so releaseLocksForClient + // would never fire). + if (granted && !replied) { + slot.lock.lock(); + try { + if (slot.heldToken == op.lockToken) { + slot.heldToken = 0; + slot.lockOwnerClientId = null; + } + } finally { + slot.lock.unlock(); + } + } + drain(slot); + } + } + + private void completeShared(KeySlot slot, Op op) { + // A reply callback may fire more than once (e.g. send failure then reply + // timeout, see NettyChannel.sendMessageWithAsyncReply); complete each op at + // most once so the readers-writer accounting is not corrupted. + if (!op.completed.compareAndSet(false, true)) { + return; + } + try { + slot.lock.lock(); + try { + slot.activeReaders--; + } finally { + slot.lock.unlock(); + } + } finally { + drain(slot); + } + } + + private void completeExclusive(KeySlot slot, Op op, RawString key, Throwable error) { + // Complete at most once: guards both a doubly-fired reply callback and the + // race between the broadcast completion and the runExclusive catch block. + if (!op.completed.compareAndSet(false, true)) { + return; + } + List> waiters; + slot.lock.lock(); + try { + slot.activeWriter = false; + if (slot.inflightInvalidate == op) { + op.acceptingCoalesce = false; + slot.inflightInvalidate = null; + } + waiters = op.coalescedWaiters; + } finally { + slot.lock.unlock(); + } + // Notify every waiter (isolating a throwing callback) and always drain, so a + // reply that fails mid-completion cannot strand the coalesced callers or leave + // the queued work on the key unresumed. + try { + RawString result = error == null ? key : null; + notifyQuietly(op.onFinish, result, error); + if (waiters != null) { + for (SimpleCallback waiter : waiters) { + notifyQuietly(waiter, result, error); + } + } + } finally { + drain(slot); + } + } + + private static void notifyQuietly(SimpleCallback callback, RawString result, Throwable error) { + if (callback == null) { + return; + } + try { + callback.onResult(result, error); + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error notifying completion callback", t); + } + } + + private void runBypass(KeySlot slot, Op op) { + if (op.mode == Mode.SHARED) { + try { + op.sharedBody.accept(() -> { + if (op.completed.compareAndSet(false, true)) { + bypassRelease(slot); + } + }); + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error running fetch (locked) for key " + op.resultKey, t); + if (op.completed.compareAndSet(false, true)) { + bypassRelease(slot); + } + } + return; + } + try { + if (op.registration != null) { + op.registration.run(); + } + Consumer broadcast = op.broadcast; + if (broadcast == null) { + bypassFinish(slot, op, op.resultKey, null); + } else { + broadcast.accept(() -> bypassFinish(slot, op, op.resultKey, null)); + } + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error running " + op.verb + " (locked) for key " + op.resultKey, t); + bypassFinish(slot, op, null, t); + } + } + + /** + * Completes a lock-bypass write op at most once: like the queued path, the reply + * callback may fire more than once, so guard onFinish and the slot bookkeeping. + */ + private void bypassFinish(KeySlot slot, Op op, RawString key, Throwable error) { + if (!op.completed.compareAndSet(false, true)) { + return; + } + try { + op.onFinish.onResult(error == null ? key : null, error); + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error replying to " + op.verb + " (locked) for key " + op.resultKey, t); + } finally { + bypassRelease(slot); + } + } + + /** + * Releases a completed lock-bypass op from the in-flight count and resumes the + * drain, so queued work that was waiting behind the bypass barrier can start once + * every bypass op has finished. + */ + private void bypassRelease(KeySlot slot) { + slot.lock.lock(); + try { + slot.activeBypass--; + } finally { + slot.lock.unlock(); + } + drain(slot); + } + + private void maybeRemove(KeySlot slot) { + slot.lock.lock(); + try { + if (!slot.draining && slot.queue.isEmpty() && slot.activeReaders == 0 && !slot.activeWriter + && slot.heldToken == 0 && slot.activeBypass == 0 && slot.inflightInvalidate == null) { + slot.removed = true; + slots.remove(slot.key, slot); + } + } finally { + slot.lock.unlock(); + } + } + + /** + * State of a single key. All mutable fields are guarded by {@link #lock}. + */ + private static final class KeySlot { + + private final RawString key; + private final ReentrantLock lock = new ReentrantLock(); + private final ArrayDeque queue = new ArrayDeque<>(); + private int activeReaders; + private boolean activeWriter; + private long heldToken; + private String lockOwnerClientId; + private int activeBypass; + private Op inflightInvalidate; + private boolean draining; + private boolean drainAgain; + private boolean removed; + + KeySlot(RawString key) { + this.key = key; + } + } + + /** + * A single queued/running operation. + */ + private static final class Op { + + private final Verb verb; + private final Mode mode; + // set for LOCK ops: the owning client, so a disconnect can cancel a queued lock + private String sourceClientId; + // completed at most once, even if the reply callback fires more than once + private final AtomicBoolean completed = new AtomicBoolean(); + + // shared (fetch) + private Consumer sharedBody; + + // exclusive write (put/load/invalidate) + private Runnable registration; + private Consumer broadcast; + private RawString resultKey; + private SimpleCallback onFinish; + private boolean coalescible; + // lazily created: most write ops never fold a predecessor nor gather waiters + private List foldedRegistrations; + private List> coalescedWaiters; + private boolean acceptingCoalesce; + + // lock + private long lockToken; + private BooleanSupplier ownerStillConnected; + private SimpleCallback lockOnFinish; + + Op(Verb verb, Mode mode) { + this.verb = verb; + this.mode = mode; + } + + // Lazy accessors, always called under the owning slot's lock. + List foldedRegistrations() { + if (foldedRegistrations == null) { + foldedRegistrations = new ArrayList<>(); + } + return foldedRegistrations; + } + + List> coalescedWaiters() { + if (coalescedWaiters == null) { + coalescedWaiters = new ArrayList<>(); + } + return coalescedWaiters; + } + } +} diff --git a/blazingcache-core/src/main/java/blazingcache/server/LockID.java b/blazingcache-core/src/main/java/blazingcache/server/LockID.java deleted file mode 100644 index 4fcbc91..0000000 --- a/blazingcache-core/src/main/java/blazingcache/server/LockID.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - Licensed to Diennea S.r.l. under one - or more contributor license agreements. See the NOTICE file - distributed with this work for additional information - regarding copyright ownership. Diennea S.r.l. licenses this file - to you under the Apache License, Version 2.0 (the - "License"); you may not use this file except in compliance - with the License. You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, - software distributed under the License is distributed on an - "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - KIND, either express or implied. See the License for the - specific language governing permissions and limitations - under the License. - - */ -package blazingcache.server; - -/** - * ID of a named lock - * - * @author enrico.olivelli - */ -public class LockID { - - public final long stamp; - - public static final LockID VALIDATED_CLIENT_PROVIDED_LOCK = new LockID(Long.MAX_VALUE); - - public LockID(long stamp) { - this.stamp = stamp; - } - - @Override - public String toString() { - return "LockID{" + "stamp=" + stamp + '}'; - } - - @Override - public int hashCode() { - int hash = 7; - hash = 13 * hash + (int) (this.stamp ^ (this.stamp >>> 32)); - return hash; - } - - @Override - public boolean equals(Object obj) { - if (this == obj) { - return true; - } - if (obj == null) { - return false; - } - if (getClass() != obj.getClass()) { - return false; - } - final LockID other = (LockID) obj; - if (this.stamp != other.stamp) { - return false; - } - return true; - } - -} diff --git a/blazingcache-core/src/main/java/blazingcache/server/PendingInvalidationsManager.java b/blazingcache-core/src/main/java/blazingcache/server/PendingInvalidationsManager.java deleted file mode 100644 index 01a3f1e..0000000 --- a/blazingcache-core/src/main/java/blazingcache/server/PendingInvalidationsManager.java +++ /dev/null @@ -1,107 +0,0 @@ -/* - Licensed to Diennea S.r.l. under one - or more contributor license agreements. See the NOTICE file - distributed with this work for additional information - regarding copyright ownership. Diennea S.r.l. licenses this file - to you under the Apache License, Version 2.0 (the - "License"); you may not use this file except in compliance - with the License. You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, - software distributed under the License is distributed on an - "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - KIND, either express or implied. See the License for the - specific language governing permissions and limitations - under the License. - - */ -package blazingcache.server; - -import blazingcache.utils.RawString; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.concurrent.ConcurrentHashMap; - -/** - * Coalesces concurrent invalidations of the same key (issue #188). - *

- * Invalidating a key is idempotent: while an invalidation of a key is in flight - * (holding the per-key write lock and broadcasting to the interested clients), any - * further invalidation request for the same key can attach to the in-flight one - * instead of queueing behind the lock for another full network round-trip. When the - * in-flight invalidation completes, every attached requester is notified with the - * same result. - *

- * This is safe because the in-flight invalidation holds the per-key write lock for - * the whole broadcast, which is mutually exclusive with fetches (read lock): no - * client can register itself for the key between the moment the interested-clients - * set is snapshotted and the moment the broadcast completes. Therefore any request - * that arrives while the broadcast is in flight is fully satisfied by it. - * - * @author diego.salvi - */ -class PendingInvalidationsManager { - - /** - * A group of invalidation requests for the same key that share a single - * broadcast. - */ - static final class PendingInvalidation { - - private final List> callbacks = - Collections.synchronizedList(new ArrayList<>()); - - private void attach(SimpleCallback callback) { - callbacks.add(callback); - } - - List> getCallbacks() { - return callbacks; - } - } - - private final ConcurrentHashMap pending = new ConcurrentHashMap<>(); - - /** - * Registers an invalidation request for a key. - * - * @return {@code true} if the caller is the owner of a NEW in-flight - * invalidation and must therefore perform the actual broadcast; {@code false} - * if the request has been coalesced into an already in-flight invalidation and - * the caller must not do anything (its callback will be invoked when the - * in-flight invalidation completes). - */ - boolean register(RawString key, SimpleCallback onFinish) { - final boolean[] owner = {false}; - pending.compute(key, (k, existing) -> { - if (existing == null) { - existing = new PendingInvalidation(); - owner[0] = true; - } - existing.attach(onFinish); - return existing; - }); - return owner[0]; - } - - /** - * Marks the in-flight invalidation of a key as completed and returns every - * callback that must be notified (the owner's plus all the coalesced ones). - * Any invalidation request arriving after this call will start a fresh - * broadcast. - */ - List> complete(RawString key) { - PendingInvalidation removed = pending.remove(key); - if (removed == null) { - return Collections.emptyList(); - } - return removed.getCallbacks(); - } - - int getNumberOfPendingInvalidations() { - return pending.size(); - } -} diff --git a/blazingcache-core/src/test/java/blazingcache/client/FetchAndInvalidateStormTest.java b/blazingcache-core/src/test/java/blazingcache/client/FetchAndInvalidateStormTest.java index d5c84c3..c56c51a 100644 --- a/blazingcache-core/src/test/java/blazingcache/client/FetchAndInvalidateStormTest.java +++ b/blazingcache-core/src/test/java/blazingcache/client/FetchAndInvalidateStormTest.java @@ -38,12 +38,12 @@ import org.junit.Test; /** - * Reproducer for issue #188: under a storm of concurrent fetch and invalidate - * operations on a single hot key, the server keeps a per-key WRITE lock held for - * the whole duration of the network round-trip (see - * {@code CacheServer.fetchEntry / invalidateKey} which acquire - * {@code KeyedLockManager.acquireWriteLockForKey} and release it only in the - * async reply callback). + * Reproducer for issue #188: historically, under a storm of concurrent fetch and + * invalidate operations on a single hot key, the server kept a per-key WRITE lock + * held for the whole duration of the network round-trip, tying up a pool thread per + * waiting operation. Coordination is now handled by {@code KeyedScheduler} (a + * non-blocking per-key event queue), so no thread ever blocks on a lock across the + * network round-trip. *

* As a consequence every operation on that key is fully serialized and each one * holds the lock for a full RTT. When many clients hit the same key at once the diff --git a/blazingcache-core/src/test/java/blazingcache/server/KeyedLockManagerLockIdTest.java b/blazingcache-core/src/test/java/blazingcache/server/KeyedLockManagerLockIdTest.java deleted file mode 100644 index 3e6f1a3..0000000 --- a/blazingcache-core/src/test/java/blazingcache/server/KeyedLockManagerLockIdTest.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - Licensed to Diennea S.r.l. under one - or more contributor license agreements. See the NOTICE file - distributed with this work for additional information - regarding copyright ownership. Diennea S.r.l. licenses this file - to you under the Apache License, Version 2.0 (the - "License"); you may not use this file except in compliance - with the License. You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, - software distributed under the License is distributed on an - "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - KIND, either express or implied. See the License for the - specific language governing permissions and limitations - under the License. - - */ -package blazingcache.server; - -import static org.junit.Assert.assertNull; -import blazingcache.utils.RawString; -import org.junit.Test; - -/** - * A malformed (non-numeric) client-provided lock id must be reported as an invalid - * lock (null) instead of throwing a NumberFormatException that would leave the - * request hanging until timeout (issue #188). - * - * @author diego.salvi - */ -public class KeyedLockManagerLockIdTest { - - private static final RawString KEY = RawString.of("k"); - - @Test - public void malformedClientProvidedLockIdIsInvalidOnWrite() { - KeyedLockManager m = new KeyedLockManager(); - assertNull(m.acquireWriteLockForKey(KEY, "client", "not-a-number")); - } - - @Test - public void malformedClientProvidedLockIdIsInvalidOnRead() { - KeyedLockManager m = new KeyedLockManager(); - assertNull(m.acquireReadLockForKey(KEY, "client", "not-a-number")); - } -} diff --git a/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java b/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java new file mode 100644 index 0000000..8259291 --- /dev/null +++ b/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java @@ -0,0 +1,490 @@ +/* + Licensed to Diennea S.r.l. under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. Diennea S.r.l. licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + + */ +package blazingcache.server; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import blazingcache.utils.RawString; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; +import org.junit.Test; + +/** + * Deterministic unit tests for {@link KeyedScheduler}. Operation completions are held + * and fired manually by the test, so scheduling/coalescing behaviour is fully + * reproducible on a single thread. + * + * @author diego.salvi + */ +public class KeyedSchedulerTest { + + private static final RawString K = RawString.of("k"); + + private final List events = new ArrayList<>(); + private final Map completions = new ConcurrentHashMap<>(); + + private Consumer broadcast(String tag) { + return onComplete -> { + events.add("bcast:" + tag); + completions.put(tag, onComplete); + }; + } + + private Consumer fetchBody(String tag) { + return onComplete -> { + events.add("fetch:" + tag); + completions.put(tag, onComplete); + }; + } + + private Runnable registration(String tag) { + return () -> events.add("reg:" + tag); + } + + private SimpleCallback onFinish(String tag) { + return (result, error) -> events.add(error == null ? "finish:" + tag : "error:" + tag); + } + + private void fire(String tag) { + Runnable r = completions.remove(tag); + assertNotNull("no pending completion for " + tag, r); + r.run(); + } + + private void put(KeyedScheduler s, String tag, String lockId) { + s.submitExclusive(K, KeyedScheduler.Verb.PUT, lockId, registration(tag), broadcast(tag), K, onFinish(tag)); + } + + private void load(KeyedScheduler s, String tag) { + // load has no broadcast + s.submitExclusive(K, KeyedScheduler.Verb.LOAD, null, registration(tag), null, K, onFinish(tag)); + } + + private void invalidate(KeyedScheduler s, String tag) { + s.submitExclusive(K, KeyedScheduler.Verb.INVALIDATE, null, registration(tag), broadcast(tag), K, onFinish(tag)); + } + + private void fetch(KeyedScheduler s, String tag) { + s.submitFetch(K, null, fetchBody(tag), () -> events.add("invalidLock:" + tag)); + } + + @Test + public void exclusiveOpsSerializeWhenFirstIsInFlight() { + KeyedScheduler s = new KeyedScheduler(); + put(s, "P0", null); // starts immediately, broadcast pending + put(s, "P1", null); // queued (P0 in flight, not in queue -> no coalescing) + assertEquals(java.util.Arrays.asList("reg:P0", "bcast:P0"), events); + assertEquals(Integer.valueOf(2), s.getLockedKeys().get(K)); // 1 active writer + 1 queued + + fire("P0"); + // P1 runs now + assertEquals(java.util.Arrays.asList("reg:P0", "bcast:P0", "finish:P0", "reg:P1", "bcast:P1"), events); + fire("P1"); + assertTrue(s.getLockedKeys().isEmpty()); + assertEquals(0, s.getNumberOfLockedKeys()); + } + + @Test + public void newerPutSuppressesQueuedPutButKeepsBothRegistrations() { + KeyedScheduler s = new KeyedScheduler(); + put(s, "P0", null); // in flight (blocker) + put(s, "P1", null); // queued + put(s, "P2", null); // folds P1 into itself (P1 removed from the queue) + // P0 in flight + P2 queued; P1 was folded (coalesced) and is not counted + assertEquals(Integer.valueOf(2), s.getLockedKeys().get(K)); + + fire("P0"); + // P1 runs suppressed (registration kept, no broadcast), then P2 broadcasts once + assertEquals(java.util.Arrays.asList("reg:P0", "bcast:P0", "finish:P0", "reg:P1", "reg:P2", "bcast:P2"), events); + fire("P2"); + // P2 completes and notifies both its own and the coalesced P1 caller + assertTrue(events.contains("finish:P2")); + assertTrue(events.contains("finish:P1")); + assertFalse("P1 must not broadcast", events.contains("bcast:P1")); + assertTrue(s.getLockedKeys().isEmpty()); + } + + @Test + public void putSuppressesQueuedLoadPreservingLoadRegistration() { + KeyedScheduler s = new KeyedScheduler(); + put(s, "P0", null); // blocker + load(s, "L1"); // queued, no broadcast + put(s, "P2", null); // dominates LOAD -> suppresses L1 + + fire("P0"); + assertEquals(java.util.Arrays.asList("reg:P0", "bcast:P0", "finish:P0", "reg:L1", "reg:P2", "bcast:P2"), events); + fire("P2"); + assertTrue(events.contains("finish:L1")); + assertTrue(events.contains("finish:P2")); + } + + @Test + public void invalidateSuppressesQueuedWritesOfAnySource() { + KeyedScheduler s = new KeyedScheduler(); + put(s, "P0", null); // blocker + put(s, "P1", null); // queued + load(s, "L1"); // queued + invalidate(s, "I2"); // dominates everything -> suppresses P1 and L1 + + fire("P0"); + // P1, L1 run suppressed (registrations kept), then I2 broadcasts once + assertEquals(java.util.Arrays.asList("reg:P0", "bcast:P0", "finish:P0", "reg:P1", "reg:L1", "reg:I2", "bcast:I2"), events); + assertFalse(events.contains("bcast:P1")); + fire("I2"); + assertTrue(events.contains("finish:P1")); + assertTrue(events.contains("finish:L1")); + assertTrue(events.contains("finish:I2")); + } + + @Test + public void coalescingStopsAtFetchBarrier() { + KeyedScheduler s = new KeyedScheduler(); + put(s, "P0", null); // blocker (in flight) + put(s, "P1", null); // queued + fetch(s, "F"); // queued behind P1 (writer in flight) + put(s, "P2", null); // scan hits fetch first -> P1 NOT suppressed + + fire("P0"); // -> P1 runs and broadcasts (not suppressed) + assertTrue(events.contains("bcast:P1")); + fire("P1"); // -> fetch F runs + assertTrue(events.contains("fetch:F")); + fire("F"); // -> P2 runs + assertTrue(events.contains("bcast:P2")); + fire("P2"); + assertTrue(s.getLockedKeys().isEmpty()); + } + + @Test + public void plainInvalidationAttachesToInFlightInvalidation() { + KeyedScheduler s = new KeyedScheduler(); + invalidate(s, "I0"); // in flight, accepting coalesce + invalidate(s, "I1"); // attaches to I0, no new broadcast, not queued + invalidate(s, "I2"); // attaches to I0 as well + + assertEquals(Integer.valueOf(1), s.getLockedKeys().get(K)); // only I0 counts + assertEquals(java.util.Arrays.asList("reg:I0", "bcast:I0"), events); + + fire("I0"); // notifies I0 + coalesced I1 + I2 + assertTrue(events.contains("finish:I0")); + assertTrue(events.contains("finish:I1")); + assertTrue(events.contains("finish:I2")); + assertFalse(events.contains("bcast:I1")); + assertFalse(events.contains("bcast:I2")); + assertTrue(s.getLockedKeys().isEmpty()); + } + + @Test + public void fetchesRunConcurrentlyAndWriterIsNotStarved() { + KeyedScheduler s = new KeyedScheduler(); + fetch(s, "F1"); // starts + fetch(s, "F2"); // starts concurrently + assertEquals(java.util.Arrays.asList("fetch:F1", "fetch:F2"), events); + assertEquals(Integer.valueOf(2), s.getLockedKeys().get(K)); + + invalidate(s, "W"); // queued: waits for the in-flight readers + fetch(s, "F3"); // fairness: queued BEHIND the waiting writer + assertFalse(events.contains("bcast:W")); + assertFalse(events.contains("fetch:F3")); + + fire("F1"); + assertFalse("writer waits for all readers", events.contains("bcast:W")); + fire("F2"); + // now readers drained -> writer runs, F3 still waiting behind it + assertTrue(events.contains("bcast:W")); + assertFalse(events.contains("fetch:F3")); + + fire("W"); + assertTrue(events.contains("fetch:F3")); + fire("F3"); + assertTrue(s.getLockedKeys().isEmpty()); + } + + @Test + public void applicationLockBypassAndQueueing() { + KeyedScheduler s = new KeyedScheduler(); + final String[] token = new String[1]; + s.submitLock(K, "owner", () -> true, + (result, error) -> { + token[0] = result; + events.add("lockFinish:" + result); + }); + assertNotNull(token[0]); + assertEquals(1, s.getNumberOfLockedKeys()); + + // op carrying the valid token bypasses the queue and runs immediately + put(s, "OWNED", token[0]); + assertTrue(events.contains("bcast:OWNED")); + fire("OWNED"); + + // malformed token -> invalid + put(s, "BAD", "not-a-number"); + assertTrue(events.contains("error:BAD")); + + // wrong token -> invalid + put(s, "WRONG", "999999"); + assertTrue(events.contains("error:WRONG")); + + // no token while locked -> queued, must NOT run yet + put(s, "WAIT", null); + assertFalse(events.contains("bcast:WAIT")); + + // unlock releases and drains the queued op + s.submitUnlock(K, Long.parseLong(token[0]), (result, error) -> events.add("unlockFinish")); + assertTrue(events.contains("unlockFinish")); + assertTrue(events.contains("bcast:WAIT")); + fire("WAIT"); + assertTrue(s.getLockedKeys().isEmpty()); + assertEquals(0, s.getNumberOfLockedKeys()); + } + + @Test + public void releaseLocksForClientReleasesHeldLockAndResumesDrain() { + KeyedScheduler s = new KeyedScheduler(); + s.submitLock(K, "owner", () -> true, (result, error) -> { + }); + put(s, "WAIT", null); // queued behind the lock + assertFalse(events.contains("bcast:WAIT")); + + s.releaseLocksForClient("owner"); + assertTrue(events.contains("bcast:WAIT")); + fire("WAIT"); + assertTrue(s.getLockedKeys().isEmpty()); + } + + /** + * A lock request still QUEUED when its client disconnects must be cancelled, not + * later granted to the gone client (leaving the key locked forever). + */ + @Test + public void releaseLocksForClientCancelsQueuedLock() { + KeyedScheduler s = new KeyedScheduler(); + final long[] owner1Token = new long[1]; + s.submitLock(K, "owner1", () -> true, + (result, error) -> owner1Token[0] = Long.parseLong(result)); + + final boolean[] cancelled = {false}; + s.submitLock(K, "owner2", () -> true, + (result, error) -> { + if (error != null) { + cancelled[0] = true; + } else { + events.add("granted2"); + } + }); + + s.releaseLocksForClient("owner2"); // owner2 disconnects while queued + assertTrue("queued lock must be cancelled", cancelled[0]); + + // owner1 unlocks: owner2's lock must NOT be granted (it was cancelled) + s.submitUnlock(K, owner1Token[0], (result, error) -> { + }); + assertFalse(events.contains("granted2")); + assertTrue(s.getLockedKeys().isEmpty()); + assertEquals(0, s.getNumberOfLockedKeys()); + } + + /** + * A token-carrying (lock-bypass) operation in flight must keep the readers-writer + * barrier: even after the lock is released, a queued exclusive op must wait for the + * in-flight bypass op to complete, so their CacheStatus effects cannot interleave. + */ + @Test + public void bypassOpHoldsBarrierUntilItCompletesAfterUnlock() { + KeyedScheduler s = new KeyedScheduler(); + final long[] token = new long[1]; + s.submitLock(K, "owner", () -> true, (result, error) -> token[0] = Long.parseLong(result)); + // owner issues a token-carrying fetch (bypass) that stays in flight + s.submitFetch(K, Long.toString(token[0]), fetchBody("BF"), () -> events.add("invalidLock:BF")); + assertTrue(events.contains("fetch:BF")); + + invalidate(s, "INV"); // queued behind the held lock + assertFalse(events.contains("bcast:INV")); + + // owner unlocks while the bypass fetch is still in flight + s.submitUnlock(K, token[0], (result, error) -> { + }); + // the invalidate must STILL wait for the in-flight bypass op + assertFalse(events.contains("bcast:INV")); + + fire("BF"); // bypass fetch completes -> barrier lifts + assertTrue(events.contains("bcast:INV")); + fire("INV"); + assertTrue(s.getLockedKeys().isEmpty()); + } + + /** + * If a LOCK request races ahead of its client's disconnect cleanup, the owner is + * already gone at grant time; the lock must be released instead of granted (else the + * key would be locked forever), and queued work must resume. + */ + @Test + public void lockForAlreadyDisconnectedOwnerIsNotGranted() { + KeyedScheduler s = new KeyedScheduler(); + final String[] outcome = {null}; + // ownerStillConnected returns false: the client disconnected before the grant + s.submitLock(K, "gone", () -> false, + (result, error) -> outcome[0] = error != null ? "aborted" : "granted"); + assertEquals("aborted", outcome[0]); + assertEquals(0, s.getNumberOfLockedKeys()); + + // work submitted afterwards must run (the key is not stuck) + put(s, "AFTER", null); + assertTrue(events.contains("bcast:AFTER")); + fire("AFTER"); + assertTrue(s.getLockedKeys().isEmpty()); + } + + /** + * Unlocking with token 0 (the "no lock held" sentinel) on an existing-but-unlocked + * slot must NOT produce a false success. + */ + @Test + public void unlockWithSentinelTokenZeroIsRejected() { + KeyedScheduler s = new KeyedScheduler(); + put(s, "P0", null); // in-flight op: the slot exists but holds no lock + final String[] outcome = {null}; + s.submitUnlock(K, 0L, (result, error) -> outcome[0] = error != null ? "error" : "ok"); + assertEquals("error", outcome[0]); + fire("P0"); + assertTrue(s.getLockedKeys().isEmpty()); + } + + /** + * If the grant reply fails to reach the client, the just-acquired lock must be + * released so the key is not stalled forever (the client never learns it holds it). + */ + @Test + public void lockGrantReplyFailureReleasesLock() { + KeyedScheduler s = new KeyedScheduler(); + s.submitLock(K, "owner", () -> true, (result, error) -> { + if (error == null) { + throw new RuntimeException("reply delivery failed"); + } + }); + assertEquals(0, s.getNumberOfLockedKeys()); + + put(s, "AFTER", null); + assertTrue(events.contains("bcast:AFTER")); + fire("AFTER"); + assertTrue(s.getLockedKeys().isEmpty()); + } + + /** + * A reply callback may fire more than once (send failure then reply timeout, see + * NettyChannel.sendMessageWithAsyncReply); a doubly-fired shared completion must + * not double-decrement the reader count and let a queued writer start while another + * fetch is still in flight. + */ + @Test + public void doubleFiredSharedCompletionIsIdempotent() { + KeyedScheduler s = new KeyedScheduler(); + fetch(s, "F1"); + fetch(s, "F2"); + assertEquals(Integer.valueOf(2), s.getLockedKeys().get(K)); + + Runnable f1 = completions.get("F1"); + f1.run(); + f1.run(); // double-fire must be a no-op + // F2 is still in flight, so exactly one reader remains + assertEquals(Integer.valueOf(1), s.getLockedKeys().get(K)); + + completions.get("F2").run(); + assertTrue(s.getLockedKeys().isEmpty()); + } + + @Test + public void doubleFiredExclusiveCompletionIsIdempotent() { + KeyedScheduler s = new KeyedScheduler(); + put(s, "P0", null); // in flight + put(s, "P1", null); // queued + + Runnable p0 = completions.get("P0"); + p0.run(); + p0.run(); // double-fire must be a no-op + // P1 must have started exactly once + assertEquals(1, java.util.Collections.frequency(events, "bcast:P1")); + + completions.get("P1").run(); + assertTrue(s.getLockedKeys().isEmpty()); + } + + /** + * The lock token must already be valid by the time the lock reply fires, so a + * client that immediately issues a token-carrying op (modelled here by submitting + * from within the reply callback) is not rejected as holding an invalid lock. + */ + @Test + public void tokenIsValidWhenLockReplyFires() { + KeyedScheduler s = new KeyedScheduler(); + s.submitLock(K, "owner", () -> true, (result, error) -> put(s, "OWNED", result)); + assertTrue(events.contains("bcast:OWNED")); + assertFalse(events.contains("error:OWNED")); + } + + @Test + public void doubleFiredBypassCompletionIsIdempotent() { + KeyedScheduler s = new KeyedScheduler(); + final String[] token = new String[1]; + s.submitLock(K, "owner", () -> true, (result, error) -> token[0] = result); + + put(s, "OWNED", token[0]); // bypasses the queue, broadcast pending + Runnable owned = completions.get("OWNED"); + owned.run(); + owned.run(); // double-fire must be a no-op + assertEquals(1, java.util.Collections.frequency(events, "finish:OWNED")); + } + + @Test + public void parseProvidedLockId() { + assertNull(KeyedScheduler.parseProvidedLockId("not-a-number")); + assertNull(KeyedScheduler.parseProvidedLockId("")); + assertEquals(Long.valueOf(42L), KeyedScheduler.parseProvidedLockId("42")); + } + + /** + * A malformed (non-numeric) client-provided lock id must be reported as an invalid + * lock instead of throwing, so the request is not left hanging until timeout + * (issue #188). The op body must not run. + */ + @Test + public void malformedProvidedLockIdIsInvalidOnExclusive() { + KeyedScheduler s = new KeyedScheduler(); + put(s, "P", "not-a-number"); + assertTrue(events.contains("error:P")); + assertFalse(events.contains("bcast:P")); + assertFalse(events.contains("reg:P")); + assertTrue(s.getLockedKeys().isEmpty()); + } + + @Test + public void malformedProvidedLockIdIsInvalidOnFetch() { + KeyedScheduler s = new KeyedScheduler(); + s.submitFetch(K, "not-a-number", fetchBody("F"), () -> events.add("invalidLock:F")); + assertTrue(events.contains("invalidLock:F")); + assertFalse(events.contains("fetch:F")); + assertTrue(s.getLockedKeys().isEmpty()); + } +} From e78da978e8c13c0b89220b0be45a808808d2dbb7 Mon Sep 17 00:00:00 2001 From: diegosalvi Date: Wed, 8 Jul 2026 14:41:26 +0200 Subject: [PATCH 2/7] Address review: lock connection-identity check and per-key invalidateByPrefix - lockKey: verify at grant time that the key is still held by the SAME connection that requested the lock (identity), not merely that the client id is connected. A LOCK racing ahead of its connection's disconnect cleanup could otherwise be granted to a new connection of a reconnected client; the grant reply would be sent on the original (dead) channel and silently dropped, stalling the key. - invalidateByPrefix: route the invalidation through the scheduler per matching key instead of a single bulk prefix removal, so each is serialized on its key slot with concurrent put/load/fetch (the bulk removal could interleave with an in-flight putEntry and resurrect an entry / desync the holder registration). Removes the now-unused sendPrefixInvalidationMessage / removeKeyByPrefixForClient. --- .../java/blazingcache/server/CacheServer.java | 53 +++++++++++-------- .../server/CacheServerSideConnection.java | 23 +------- .../java/blazingcache/server/CacheStatus.java | 31 ----------- 3 files changed, 31 insertions(+), 76 deletions(-) diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java index e654f9c..befae32 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java @@ -372,12 +372,15 @@ private void broadcastInvalidation(RawString key, String sourceClientId, SimpleC }); } - public void lockKey(RawString key, String sourceClientId, SimpleCallback onFinish) { - // The scheduler owns the lock lifecycle. It checks, at grant time, that the - // owning client is still connected, so a LOCK request that raced ahead of the - // client's disconnect cleanup is not granted a lock that would never be released. + public void lockKey(RawString key, String sourceClientId, CacheServerSideConnection connection, SimpleCallback onFinish) { + // The scheduler owns the lock lifecycle. At grant time it verifies the lock is + // still owned by the SAME connection that requested it, not merely that the + // client id is still connected: a LOCK that raced ahead of its connection's + // disconnect cleanup must not be granted to a *new* connection of the same + // client that reconnected in the meantime — the grant reply would be sent on the + // original (dead) channel and silently dropped, stalling the key forever. scheduler.submitLock(key, sourceClientId, - () -> acceptor.getActualConnectionFromClient(sourceClientId) != null, + () -> acceptor.getActualConnectionFromClient(sourceClientId) == connection, onFinish); } @@ -493,29 +496,33 @@ public void fetchEntry(RawString key, String clientId, String clientProvidedLock public void invalidateByPrefix(RawString prefix, String sourceClientId, SimpleCallback onFinish) { Runnable action = () -> { - // LOCKS ?? - Set clients = cacheStatus.getAllClientsWithListener(); - if (sourceClientId != null) { - clients.remove(sourceClientId); + // Enumerate the keys matching the prefix and run a per-key invalidation + // through the scheduler, so each one is serialized (on its own key slot) with + // any concurrent put/load/fetch on that key. A single bulk prefix removal + // (the previous approach) did not go through the scheduler and could + // interleave with an in-flight putEntry on a matching key: the client would + // re-store the entry after dropping it while the server had already removed + // its registration, leaving the client holding stale data the server no + // longer knows about. + List keys = new ArrayList<>(); + for (RawString key : cacheStatus.getKeys()) { + if (key.startsWith(prefix)) { + keys.add(key); + } } - if (clients.isEmpty()) { + if (keys.isEmpty()) { onFinish.onResult(prefix, null); return; } - BroadcastRequestStatus invalidation = new BroadcastRequestStatus("invalidateByPrefix " + prefix + " from " + sourceClientId + " started at " + new java.sql.Timestamp(System.currentTimeMillis()), clients, onFinish, (clientId, error) -> { - cacheStatus.removeKeyByPrefixForClient(prefix, clientId); - }); - - networkRequestsStatusMonitor.register(invalidation); - clients.forEach((clientId) -> { - CacheServerSideConnection connection = acceptor.getActualConnectionFromClient(clientId); - if (connection == null) { - LOGGER.log(Level.SEVERE, "client " + clientId + " not connected, considering prefix " + prefix + " invalidated"); - invalidation.clientDone(clientId); - } else { - connection.sendPrefixInvalidationMessage(sourceClientId, prefix, invalidation); + AtomicInteger remaining = new AtomicInteger(keys.size()); + SimpleCallback perKey = (result, error) -> { + if (remaining.decrementAndGet() == 0) { + onFinish.onResult(prefix, null); } - }); + }; + for (RawString key : keys) { + invalidateKey(key, sourceClientId, null, perKey); + } }; executeOnHandler("invalidateByPrefix " + prefix, action); } diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java index c82aedc..ee8e548 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java @@ -244,7 +244,7 @@ public void messageReceived(Message message) { } RawString key = (RawString) message.parameters.get("key"); server.addPendingOperations(1); - server.lockKey(key, clientId, new SimpleCallback() { + server.lockKey(key, clientId, this, new SimpleCallback() { @Override public void onResult(String result, Throwable error) { server.addPendingOperations(-1); @@ -554,27 +554,6 @@ public void replyReceived(Message originalMessage, Message message, Throwable er } - void sendPrefixInvalidationMessage(String sourceClientId, RawString prefix, BroadcastRequestStatus invalidation) { - Channel _channel = channel; - if (_channel == null || !_channel.isValid()) { - // not connected, quindi cache vuota - invalidation.clientDone(clientId); - return; - } - _channel.sendMessageWithAsyncReply(Message.INVALIDATE_BY_PREFIX(sourceClientId, prefix), server.getSlowClientTimeout(), new ReplyCallback() { - - @Override - public void replyReceived(Message originalMessage, Message message, Throwable error) { - LOGGER.log(Level.FINEST, clientId + " answered to invalidateByPrefix " + prefix + ": " + message + ", " + error); - if (error != null) { - error.printStackTrace(); - } - // in ogni caso il client ha finito - invalidation.clientDone(clientId); - } - }); - } - void sendFetchKeyMessage(String remoteClientId, RawString key, SimpleCallback onFinish) { Channel _channel = channel; if (_channel == null || !_channel.isValid()) { diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java b/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java index 9564652..a8b9a57 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java @@ -156,37 +156,6 @@ public void removeKeyForClient(RawString key, String client) { LOGGER.log(Level.FINEST, "removeKeyForClient key={0} client={1} -> keysForClient {2}", new Object[]{key, client, keysForClient}); } - public void removeKeyByPrefixForClient(RawString prefix, String client) { - LOGGER.log(Level.FINEST, "removeKeyByPrefixForClient prefix={0} client={1}", new Object[]{prefix, client}); - lock.writeLock().lock(); - try { - - Set keys = keysForClient.get(client); - Set selectedKeys; - if (keys != null) { - selectedKeys = keys.stream().filter(key -> key.startsWith(prefix)).collect(Collectors.toSet()); - keys.removeAll(selectedKeys); - if (keys.isEmpty()) { - keysForClient.remove(client); - } - } else { - selectedKeys = Collections.emptySet(); - } - for (RawString key : selectedKeys) { - Set clients = clientsForKey.get(key); - if (clients != null) { - clients.remove(client); - if (clients.isEmpty()) { - clientsForKey.remove(key); - entryExpireTime.remove(key); - } - } - } - } finally { - lock.writeLock().unlock(); - } - } - /** * Removes all the key listeners of a disconnected client. * From 31f9bb954fd0dad5c56035919cff779f9be6d59e Mon Sep 17 00:00:00 2001 From: diegosalvi Date: Wed, 8 Jul 2026 15:33:58 +0200 Subject: [PATCH 3/7] invalidateByPrefix: coalesce client notification into a single broadcast Keep the per-key scheduler serialization that makes prefix invalidation coherent with concurrent per-key operations, but restore the single INVALIDATE_BY_PREFIX message per client so the network fan-out stays O(clients) instead of O(matching keys) (important for large prefixes). - Phase 1: for each matching key, an exclusive scheduler op clears the holder registrations (CacheStatus.removeKey) with no network, serialized on that key's slot with any concurrent put/load/fetch. - Phase 2, after all phase-1 removals complete: a single prefix broadcast per client tells each one to drop its matching keys locally. The client set is captured before phase 1 (the removals would otherwise hide the very clients that held the matching keys). Running phase 2 after phase 1 also orders any PUT_ENTRY from a concurrent put before this invalidation on the client's channel. --- .../java/blazingcache/server/CacheServer.java | 55 +++++++++++++++---- .../server/CacheServerSideConnection.java | 21 +++++++ .../java/blazingcache/server/CacheStatus.java | 28 ++++++++++ 3 files changed, 92 insertions(+), 12 deletions(-) diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java index befae32..8f76c9d 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java @@ -496,14 +496,44 @@ public void fetchEntry(RawString key, String clientId, String clientProvidedLock public void invalidateByPrefix(RawString prefix, String sourceClientId, SimpleCallback onFinish) { Runnable action = () -> { - // Enumerate the keys matching the prefix and run a per-key invalidation - // through the scheduler, so each one is serialized (on its own key slot) with - // any concurrent put/load/fetch on that key. A single bulk prefix removal - // (the previous approach) did not go through the scheduler and could - // interleave with an in-flight putEntry on a matching key: the client would - // re-store the entry after dropping it while the server had already removed - // its registration, leaving the client holding stale data the server no - // longer knows about. + // Capture the clients to notify BEFORE phase 1: the per-key removals clear the + // holder registrations, so computing this set afterwards would miss the very + // clients that held the matching keys. + Set clientsToNotify = cacheStatus.getAllClientsWithListener(); + if (sourceClientId != null) { + clientsToNotify.remove(sourceClientId); + } + // Phase 2: notify every client with a SINGLE prefix broadcast (as before), so + // the message fan-out stays O(clients), not O(matching keys). Runs only after + // the server-side removals of phase 1 have completed, so that any PUT_ENTRY a + // concurrent put sent to a client is ordered before this invalidation on that + // client's channel. + Runnable broadcastPrefixToClients = () -> { + if (clientsToNotify.isEmpty()) { + onFinish.onResult(prefix, null); + return; + } + // holder registrations were already cleared per key in phase 1, so this + // broadcast only tells the clients to drop the matching keys locally + BroadcastRequestStatus invalidation = new BroadcastRequestStatus("invalidateByPrefix " + prefix + " from " + sourceClientId + " started at " + new java.sql.Timestamp(System.currentTimeMillis()), clientsToNotify, onFinish, null); + networkRequestsStatusMonitor.register(invalidation); + clientsToNotify.forEach((clientId) -> { + CacheServerSideConnection connection = acceptor.getActualConnectionFromClient(clientId); + if (connection == null) { + LOGGER.log(Level.SEVERE, "client " + clientId + " not connected, considering prefix " + prefix + " invalidated"); + invalidation.clientDone(clientId); + } else { + connection.sendPrefixInvalidationMessage(sourceClientId, prefix, invalidation); + } + }); + }; + + // Phase 1: clear each matching key on its own scheduler slot (no network), so + // the server-side removal is serialized with any concurrent put/load/fetch on + // that key. A single bulk prefix removal (the previous approach) did not go + // through the scheduler and could interleave with an in-flight putEntry on a + // matching key, leaving the client holding an entry the server no longer knows + // about (stale forever). List keys = new ArrayList<>(); for (RawString key : cacheStatus.getKeys()) { if (key.startsWith(prefix)) { @@ -511,17 +541,18 @@ public void invalidateByPrefix(RawString prefix, String sourceClientId, SimpleCa } } if (keys.isEmpty()) { - onFinish.onResult(prefix, null); + broadcastPrefixToClients.run(); return; } AtomicInteger remaining = new AtomicInteger(keys.size()); - SimpleCallback perKey = (result, error) -> { + SimpleCallback perKeyRemoved = (result, error) -> { if (remaining.decrementAndGet() == 0) { - onFinish.onResult(prefix, null); + broadcastPrefixToClients.run(); } }; for (RawString key : keys) { - invalidateKey(key, sourceClientId, null, perKey); + scheduler.submitExclusive(key, KeyedScheduler.Verb.INVALIDATE, null, + () -> cacheStatus.removeKey(key), null, key, perKeyRemoved); } }; executeOnHandler("invalidateByPrefix " + prefix, action); diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java index ee8e548..6803628 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java @@ -554,6 +554,27 @@ public void replyReceived(Message originalMessage, Message message, Throwable er } + void sendPrefixInvalidationMessage(String sourceClientId, RawString prefix, BroadcastRequestStatus invalidation) { + Channel _channel = channel; + if (_channel == null || !_channel.isValid()) { + // not connected, quindi cache vuota + invalidation.clientDone(clientId); + return; + } + _channel.sendMessageWithAsyncReply(Message.INVALIDATE_BY_PREFIX(sourceClientId, prefix), server.getSlowClientTimeout(), new ReplyCallback() { + + @Override + public void replyReceived(Message originalMessage, Message message, Throwable error) { + LOGGER.log(Level.FINEST, clientId + " answered to invalidateByPrefix " + prefix + ": " + message + ", " + error); + if (error != null) { + error.printStackTrace(); + } + // in ogni caso il client ha finito + invalidation.clientDone(clientId); + } + }); + } + void sendFetchKeyMessage(String remoteClientId, RawString key, SimpleCallback onFinish) { Channel _channel = channel; if (_channel == null || !_channel.isValid()) { diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java b/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java index a8b9a57..740225a 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java @@ -156,6 +156,34 @@ public void removeKeyForClient(RawString key, String client) { LOGGER.log(Level.FINEST, "removeKeyForClient key={0} client={1} -> keysForClient {2}", new Object[]{key, client, keysForClient}); } + /** + * Removes a key entirely, dropping the registration of every client that held it. + * Used by prefix invalidation, which clears each matching key on its own scheduler + * slot (serialized with concurrent per-key operations) before the clients are + * notified with a single prefix broadcast. + */ + void removeKey(RawString key) { + LOGGER.log(Level.FINEST, "removeKey key={0}", key); + lock.writeLock().lock(); + try { + Set clients = clientsForKey.remove(key); + entryExpireTime.remove(key); + if (clients != null) { + for (String client : clients) { + Set keys = keysForClient.get(client); + if (keys != null) { + keys.remove(key); + if (keys.isEmpty()) { + keysForClient.remove(client); + } + } + } + } + } finally { + lock.writeLock().unlock(); + } + } + /** * Removes all the key listeners of a disconnected client. * From 3b2bcb7004409a007c4b6fcb649d4fc1801a1346 Mon Sep 17 00:00:00 2001 From: diegosalvi Date: Wed, 8 Jul 2026 16:55:28 +0200 Subject: [PATCH 4/7] Document the benign put/load-vs-invalidateByPrefix race between the two phases --- .../main/java/blazingcache/server/CacheServer.java | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java index 8f76c9d..0e8560c 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java @@ -534,6 +534,20 @@ public void invalidateByPrefix(RawString prefix, String sourceClientId, SimpleCa // through the scheduler and could interleave with an in-flight putEntry on a // matching key, leaving the client holding an entry the server no longer knows // about (stale forever). + // + // Concurrency note - a put/load on a matching key that arrives AFTER its phase-1 + // removal but before the phase-2 broadcast re-registers that client, which then + // drops the key on the phase-2 broadcast: the server is left transiently + // believing the client still holds a key it dropped. This divergence is benign + // and self-healing (never the reverse "client holds it, server does not know" + // that would serve stale data): the next invalidate/put on the key, or the + // client disconnect, reconciles it, and a fetch routed to that client simply + // misses. The dangerous direction is prevented by the per-key serialization + // above plus running phase 2 only after phase 1, which orders any PUT_ENTRY a + // concurrent put sent to a client before this invalidation on its channel. + // Removing even the benign residual would require either a per-client bulk + // removal (reintroducing the race) or a per-key client broadcast (the O(keys) + // fan-out this design avoids), so it is accepted on purpose. List keys = new ArrayList<>(); for (RawString key : cacheStatus.getKeys()) { if (key.startsWith(prefix)) { From dd1fca807c5c2c1fa7ca5083dc6bff092f5055a5 Mon Sep 17 00:00:00 2001 From: diegosalvi Date: Wed, 15 Jul 2026 10:17:41 +0200 Subject: [PATCH 5/7] Address second review round - KeyedScheduler.submitFetch: set op.resultKey so fetch (and lock-bypass) error logs no longer print "for key null". - lock ownership tracked by CONNECTION identity, not client id: the slot keeps the owning connection id and the disconnect release (releaseLocksForConnection) only releases/cancels that connection's locks. This closes the release-side mirror of the grant-side race: a late cleanup of a client's dead connection no longer releases a lock a new connection of the same client legitimately acquired after reconnecting. - runLock: reword the granted-but-not-replied guard comment to make clear it only covers a reply callback that throws; a reply silently dropped on a dead channel is handled by the connection-identity check plus releaseLocksForConnection. - invalidateByPrefix: document that, unlike the legacy path (which ignored locks), it now waits for held application locks (no timeout) and in-flight broadcasts on the matching keys. - tests: KeyedSchedulerTest.releaseLocksForConnectionIgnoresOtherConnections (identity release) and InvalidateByPrefixResurrectionTest (no client-holds/server-unaware divergence under concurrent put + prefix invalidation). --- .../java/blazingcache/server/CacheServer.java | 35 +++--- .../server/CacheServerSideConnection.java | 2 +- .../blazingcache/server/KeyedScheduler.java | 90 ++++++++------- .../InvalidateByPrefixResurrectionTest.java | 103 ++++++++++++++++++ .../server/KeyedSchedulerTest.java | 49 ++++++--- 5 files changed, 213 insertions(+), 66 deletions(-) create mode 100644 blazingcache-core/src/test/java/blazingcache/InvalidateByPrefixResurrectionTest.java diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java index 0e8560c..166e236 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java @@ -373,13 +373,14 @@ private void broadcastInvalidation(RawString key, String sourceClientId, SimpleC } public void lockKey(RawString key, String sourceClientId, CacheServerSideConnection connection, SimpleCallback onFinish) { - // The scheduler owns the lock lifecycle. At grant time it verifies the lock is - // still owned by the SAME connection that requested it, not merely that the - // client id is still connected: a LOCK that raced ahead of its connection's - // disconnect cleanup must not be granted to a *new* connection of the same - // client that reconnected in the meantime — the grant reply would be sent on the - // original (dead) channel and silently dropped, stalling the key forever. - scheduler.submitLock(key, sourceClientId, + // The scheduler owns the lock lifecycle and ties it to this exact connection (by + // id), not just to the client id. At grant time it verifies the lock is still + // owned by the SAME connection that requested it, and on disconnect it releases + // only the locks of that connection. This closes the acquire-vs-disconnect race + // in both directions when a client dies and reconnects with the same id on a new + // connection: the orphaned grant is not handed to the new connection, and a late + // cleanup of the dead connection does not release the new connection's lock. + scheduler.submitLock(key, connection.getConnectionId(), () -> acceptor.getActualConnectionFromClient(sourceClientId) == connection, onFinish); } @@ -548,6 +549,13 @@ public void invalidateByPrefix(RawString prefix, String sourceClientId, SimpleCa // Removing even the benign residual would require either a per-client bulk // removal (reintroducing the race) or a per-key client broadcast (the O(keys) // fan-out this design avoids), so it is accepted on purpose. + // + // Behavioural change vs. the legacy prefix invalidation (which ignored locks + // and per-key coordination entirely): because phase 1 goes through the + // scheduler, a matching key that is currently held by an application lock + // (which has NO timeout) or has an invalidate/put broadcast in flight makes + // this prefix invalidation WAIT until that lock is released / that broadcast + // completes, before its clients are notified. List keys = new ArrayList<>(); for (RawString key : cacheStatus.getKeys()) { if (key.startsWith(prefix)) { @@ -580,13 +588,14 @@ private void executeOnHandler(String name, Runnable runnable) { } } - void clientDisconnected(String clientId) { + void clientDisconnected(String clientId, long connectionId) { int count = cacheStatus.removeClientListeners(clientId); - LOGGER.log(Level.SEVERE, "client " + clientId + " disconnected, removed " + count + " key listeners"); - // Release every application lock the client held or was queued for. This is - // driven entirely by the scheduler (which owns the lock ownership), so a lock - // being acquired or still queued when the client dropped is not left dangling. - scheduler.releaseLocksForClient(clientId); + LOGGER.log(Level.SEVERE, "client " + clientId + " (connection " + connectionId + ") disconnected, removed " + count + " key listeners"); + // Release every application lock held or queued by THIS connection. Keying the + // release on the connection id (not the client id) means a late cleanup of a dead + // connection cannot release a lock that a new connection of the same client + // acquired after reconnecting. + scheduler.releaseLocksForConnection(connectionId); } public long getCurrentTimestamp() { diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java index 6803628..bd157c0 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java @@ -442,7 +442,7 @@ public void channelClosed() { } channel = null; server.getAcceptor().connectionClosed(this); - server.clientDisconnected(clientId); + server.clientDisconnected(clientId, connectionId); } void answerConnectionNotAcceptedAndClose(Message connectionRequestMessage, Throwable ex diff --git a/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java b/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java index 29dfa20..b227969 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java +++ b/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java @@ -113,6 +113,7 @@ public void submitFetch(RawString key, String providedLockId, Consumer body, Runnable onInvalidLock) { Op op = new Op(Verb.FETCH, Mode.SHARED); op.sharedBody = body; + op.resultKey = key; // only used for diagnostics (e.g. error logging) submit(key, op, providedLockId, onInvalidLock); } @@ -146,19 +147,21 @@ public void submitExclusive(RawString key, Verb verb, String providedLockId, * Acquires an application lock on the key. When the lock op reaches the head of the * queue a fresh monotonic token is published atomically as the held token; the slot * then stays "held" (no further queued op is drained) until {@link #submitUnlock} - * (or {@link #releaseLocksForClient} on disconnect) releases it. Operations carrying - * the matching token bypass the queue in the meantime. + * (or {@link #releaseLocksForConnection} on disconnect) releases it. Operations + * carrying the matching token bypass the queue in the meantime. *

- * {@code ownerStillConnected} is checked at grant time: if the owning client has - * disconnected by then (its LOCK request having raced ahead of the disconnect - * cleanup), the lock is immediately released instead of being granted to a gone - * client, closing the acquire-vs-disconnect window. + * The lock is owned by a specific connection ({@code ownerConnectionId}), not merely + * by a client id, so both the grant-time liveness check ({@code ownerStillConnected}) + * and the disconnect release ({@link #releaseLocksForConnection}) are tied to the + * exact connection that acquired it. This closes the acquire-vs-disconnect race in + * both directions when a client dies and reconnects with the same id on a new + * connection. */ - public void submitLock(RawString key, String clientId, + public void submitLock(RawString key, long ownerConnectionId, BooleanSupplier ownerStillConnected, SimpleCallback onFinish) { long token = tokenGenerator.incrementAndGet(); Op op = new Op(Verb.LOCK, Mode.EXCLUSIVE); - op.sourceClientId = clientId; + op.ownerConnectionId = ownerConnectionId; op.lockToken = token; op.ownerStillConnected = ownerStillConnected; op.lockOnFinish = onFinish; @@ -193,7 +196,7 @@ public void submitUnlock(RawString key, long token, SimpleCallback onFin // unlocked key does not produce a false success if (token != 0 && slot.heldToken == token) { slot.heldToken = 0; - slot.lockOwnerClientId = null; + slot.lockOwnerConnectionId = 0; released = true; } } finally { @@ -217,28 +220,31 @@ public void submitUnlock(RawString key, long token, SimpleCallback onFin } /** - * Releases every application lock owned by a disconnected client and cancels its - * still-queued lock requests, then resumes the drain of the affected keys. This is - * the single, scheduler-driven release path used on disconnect: because the lock - * ownership ({@code heldToken} + {@code lockOwnerClientId}) is published atomically - * at dequeue, there is no window in which a held or being-acquired lock is invisible - * here, and queued lock requests that never ran are cancelled rather than later - * granted to the gone client. + * Releases every application lock owned by a disconnected CONNECTION and cancels the + * lock requests it still had queued, then resumes the drain of the affected keys. + *

+ * Ownership is tracked by connection identity ({@code ownerConnectionId}), not by + * client id, so a late cleanup of a dead connection cannot release a lock that a new + * connection of the same client legitimately acquired after reconnecting. This is the + * single, scheduler-driven release path used on disconnect: because the lock + * ownership is published atomically at dequeue, there is no window in which a held or + * being-acquired lock is invisible here, and queued lock requests that never ran are + * cancelled rather than later granted to the gone connection. */ - public void releaseLocksForClient(String clientId) { + public void releaseLocksForConnection(long connectionId) { slots.forEach((key, slot) -> { List> cancelledLocks = null; boolean changed = false; slot.lock.lock(); try { - if (slot.heldToken != 0 && clientId.equals(slot.lockOwnerClientId)) { + if (slot.heldToken != 0 && slot.lockOwnerConnectionId == connectionId) { slot.heldToken = 0; - slot.lockOwnerClientId = null; + slot.lockOwnerConnectionId = 0; changed = true; } for (Iterator it = slot.queue.iterator(); it.hasNext();) { Op queued = it.next(); - if (queued.verb == Verb.LOCK && clientId.equals(queued.sourceClientId)) { + if (queued.verb == Verb.LOCK && queued.ownerConnectionId == connectionId) { it.remove(); if (cancelledLocks == null) { cancelledLocks = new ArrayList<>(); @@ -253,7 +259,7 @@ public void releaseLocksForClient(String clientId) { if (cancelledLocks != null) { for (SimpleCallback cancelled : cancelledLocks) { try { - cancelled.onResult(null, new Exception("client " + clientId + cancelled.onResult(null, new Exception("connection " + connectionId + " disconnected while waiting for the lock on " + key)); } catch (Throwable t) { LOGGER.log(Level.SEVERE, "error cancelling queued lock for key " + key, t); @@ -486,7 +492,7 @@ private void selectRunnable(KeySlot slot, List toStart) { // race ahead of the acquisition. No activeWriter is taken, so the // key is not double-counted while runLock records it. slot.heldToken = head.lockToken; - slot.lockOwnerClientId = head.sourceClientId; + slot.lockOwnerConnectionId = head.ownerConnectionId; } else { slot.activeWriter = true; if (head.verb == Verb.INVALIDATE && head.coalescible) { @@ -547,14 +553,15 @@ private void runExclusive(KeySlot slot, Op op) { } private void runLock(KeySlot slot, Op op) { - // heldToken + lockOwnerClientId were published at dequeue (selectRunnable). Only - // keep the lock if we still hold it AND the owning client is still connected: a - // concurrent disconnect may already have released it, or the LOCK request may - // have raced ahead of the disconnect cleanup (which runs after the connection is - // removed), in which case the owner is already gone. - // evaluate liveness outside the slot lock (it calls into the acceptor); the tiny - // window between this check and the grant is backstopped by releaseLocksForClient, - // which runs after the connection is removed and finds the published heldToken. + // heldToken + lockOwnerConnectionId were published at dequeue (selectRunnable). + // Only keep the lock if we still hold it AND the owning connection is still the + // current one for the client: a concurrent disconnect may already have released + // it, or the LOCK request may have raced ahead of the disconnect cleanup (which + // runs after the connection is removed), in which case the owner is already gone. + // Evaluate liveness outside the slot lock (it calls into the acceptor); the tiny + // window between this check and the grant is backstopped by + // releaseLocksForConnection, which runs after the connection is removed and finds + // the published heldToken. boolean ownerConnected = op.ownerStillConnected == null || op.ownerStillConnected.getAsBoolean(); boolean granted = false; slot.lock.lock(); @@ -565,7 +572,7 @@ private void runLock(KeySlot slot, Op op) { } else { // release the lock we just published: the owner disconnected slot.heldToken = 0; - slot.lockOwnerClientId = null; + slot.lockOwnerConnectionId = 0; } } } finally { @@ -583,16 +590,19 @@ private void runLock(KeySlot slot, Op op) { } catch (Throwable t) { LOGGER.log(Level.SEVERE, "error replying to lockKey for key " + slot.key, t); } finally { - // If the grant reply failed to reach the client, release the just-acquired - // lock: the client does not know it holds it, so leaving heldToken set would - // stall the key forever (the client stays connected, so releaseLocksForClient - // would never fire). + // This guard only covers the case where the reply callback itself THROWS + // (e.g. an encoder failure): then the owner cannot know it holds the lock, so + // we release it to avoid stalling the key. The other failure mode - the reply + // being silently dropped on an already-dead channel (NettyChannel discards it + // without throwing) - does NOT reach here; that case is handled by the + // connection-identity liveness check above plus releaseLocksForConnection on + // the disconnect. if (granted && !replied) { slot.lock.lock(); try { if (slot.heldToken == op.lockToken) { slot.heldToken = 0; - slot.lockOwnerClientId = null; + slot.lockOwnerConnectionId = 0; } } finally { slot.lock.unlock(); @@ -754,7 +764,7 @@ private static final class KeySlot { private int activeReaders; private boolean activeWriter; private long heldToken; - private String lockOwnerClientId; + private long lockOwnerConnectionId; private int activeBypass; private Op inflightInvalidate; private boolean draining; @@ -773,8 +783,10 @@ private static final class Op { private final Verb verb; private final Mode mode; - // set for LOCK ops: the owning client, so a disconnect can cancel a queued lock - private String sourceClientId; + // set for LOCK ops: the id of the owning connection, so the lock is tied to that + // exact connection (a disconnect releases/cancels only its own locks, even if the + // client reconnected with the same id on a new connection) + private long ownerConnectionId; // completed at most once, even if the reply callback fires more than once private final AtomicBoolean completed = new AtomicBoolean(); diff --git a/blazingcache-core/src/test/java/blazingcache/InvalidateByPrefixResurrectionTest.java b/blazingcache-core/src/test/java/blazingcache/InvalidateByPrefixResurrectionTest.java new file mode 100644 index 0000000..2ea6da4 --- /dev/null +++ b/blazingcache-core/src/test/java/blazingcache/InvalidateByPrefixResurrectionTest.java @@ -0,0 +1,103 @@ +/* + Licensed to Diennea S.r.l. under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. Diennea S.r.l. licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + + */ +package blazingcache; + +import static org.junit.Assert.assertTrue; +import blazingcache.client.CacheClient; +import blazingcache.client.EntryHandle; +import blazingcache.network.ServerHostData; +import blazingcache.network.netty.NettyCacheServerLocator; +import blazingcache.server.CacheServer; +import blazingcache.utils.RawString; +import java.nio.charset.StandardCharsets; +import org.junit.Test; + +/** + * Reproducer for the coherence bug that motivated the two-phase {@code invalidateByPrefix} + * rework: a prefix invalidation must not leave a client holding an entry the server no + * longer knows about (which would then miss future invalidations and serve stale data). + *

+ * Each round a holder client owns a key, another client overwrites it with a putEntry + * (which the server broadcasts to the holder) while a third client invalidates by prefix + * concurrently. The invariant checked after every round is the safe direction: if the + * holder still has the key locally, the server must know it holds it. + */ +public class InvalidateByPrefixResurrectionTest { + + private static final String PREFIX = "k"; + private static final RawString KEY = RawString.of("k1"); + private static final int ROUNDS = 300; + + @Test(timeout = 120000) + public void prefixInvalidationDoesNotResurrectEntries() throws Exception { + byte[] data = "v".getBytes(StandardCharsets.UTF_8); + byte[] data2 = "v2".getBytes(StandardCharsets.UTF_8); + + ServerHostData serverHostData = new ServerHostData("localhost", 1234, "test", false, null); + try (CacheServer cacheServer = new CacheServer("ciao", serverHostData)) { + cacheServer.start(); + try (CacheClient holder = new CacheClient("holder", "ciao", new NettyCacheServerLocator(serverHostData)); + CacheClient putter = new CacheClient("putter", "ciao", new NettyCacheServerLocator(serverHostData)); + CacheClient invalidator = new CacheClient("invalidator", "ciao", new NettyCacheServerLocator(serverHostData))) { + holder.start(); + putter.start(); + invalidator.start(); + assertTrue(holder.waitForConnection(10000)); + assertTrue(putter.waitForConnection(10000)); + assertTrue(invalidator.waitForConnection(10000)); + + for (int i = 0; i < ROUNDS; i++) { + // holder owns the key (server registers "holder") + holder.put(KEY.toString(), data, 0); + + // concurrently: putter overwrites the key (the server broadcasts a + // PUT_ENTRY to the holder) while a third client invalidates by prefix + Thread put = new Thread(() -> { + try { + putter.put(KEY.toString(), data2, 0); + } catch (Exception ignore) { + } + }); + Thread inv = new Thread(() -> { + try { + invalidator.invalidateByPrefix(PREFIX); + } catch (Exception ignore) { + } + }); + put.start(); + inv.start(); + put.join(); + inv.join(); + + // Safe-direction invariant: never "client holds it, server does not know". + EntryHandle local = holder.get(KEY.toString()); + if (local != null) { + local.close(); + boolean serverKnows = cacheServer.getCacheStatus() + .getKeysForClient(holder.getClientId()).contains(KEY); + assertTrue("resurrection at round " + i + + ": holder has the entry but the server does not know it does", + serverKnows); + } + } + } + } + } +} diff --git a/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java b/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java index 8259291..0c79e6b 100644 --- a/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java +++ b/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java @@ -226,7 +226,7 @@ public void fetchesRunConcurrentlyAndWriterIsNotStarved() { public void applicationLockBypassAndQueueing() { KeyedScheduler s = new KeyedScheduler(); final String[] token = new String[1]; - s.submitLock(K, "owner", () -> true, + s.submitLock(K, 1L, () -> true, (result, error) -> { token[0] = result; events.add("lockFinish:" + result); @@ -261,14 +261,37 @@ public void applicationLockBypassAndQueueing() { } @Test - public void releaseLocksForClientReleasesHeldLockAndResumesDrain() { + public void releaseLocksForConnectionReleasesHeldLockAndResumesDrain() { KeyedScheduler s = new KeyedScheduler(); - s.submitLock(K, "owner", () -> true, (result, error) -> { + s.submitLock(K, 1L, () -> true, (result, error) -> { }); put(s, "WAIT", null); // queued behind the lock assertFalse(events.contains("bcast:WAIT")); - s.releaseLocksForClient("owner"); + s.releaseLocksForConnection(1L); + assertTrue(events.contains("bcast:WAIT")); + fire("WAIT"); + assertTrue(s.getLockedKeys().isEmpty()); + } + + /** + * The lock is owned by a connection, not just a client id: a disconnect of a + * DIFFERENT connection (e.g. the client's old, dead connection after it reconnected) + * must not release the lock the current connection legitimately holds. + */ + @Test + public void releaseLocksForConnectionIgnoresOtherConnections() { + KeyedScheduler s = new KeyedScheduler(); + s.submitLock(K, 1L, () -> true, (result, error) -> { + }); // held by connection 1 + put(s, "WAIT", null); // queued behind the lock + assertFalse(events.contains("bcast:WAIT")); + + s.releaseLocksForConnection(2L); // a different (old/dead) connection disconnects + assertFalse("another connection's disconnect must not release the lock", + events.contains("bcast:WAIT")); + + s.releaseLocksForConnection(1L); // the owning connection disconnects assertTrue(events.contains("bcast:WAIT")); fire("WAIT"); assertTrue(s.getLockedKeys().isEmpty()); @@ -279,14 +302,14 @@ public void releaseLocksForClientReleasesHeldLockAndResumesDrain() { * later granted to the gone client (leaving the key locked forever). */ @Test - public void releaseLocksForClientCancelsQueuedLock() { + public void releaseLocksForConnectionCancelsQueuedLock() { KeyedScheduler s = new KeyedScheduler(); final long[] owner1Token = new long[1]; - s.submitLock(K, "owner1", () -> true, + s.submitLock(K, 1L, () -> true, (result, error) -> owner1Token[0] = Long.parseLong(result)); final boolean[] cancelled = {false}; - s.submitLock(K, "owner2", () -> true, + s.submitLock(K, 2L, () -> true, (result, error) -> { if (error != null) { cancelled[0] = true; @@ -295,7 +318,7 @@ public void releaseLocksForClientCancelsQueuedLock() { } }); - s.releaseLocksForClient("owner2"); // owner2 disconnects while queued + s.releaseLocksForConnection(2L); // owner2 disconnects while queued assertTrue("queued lock must be cancelled", cancelled[0]); // owner1 unlocks: owner2's lock must NOT be granted (it was cancelled) @@ -315,7 +338,7 @@ public void releaseLocksForClientCancelsQueuedLock() { public void bypassOpHoldsBarrierUntilItCompletesAfterUnlock() { KeyedScheduler s = new KeyedScheduler(); final long[] token = new long[1]; - s.submitLock(K, "owner", () -> true, (result, error) -> token[0] = Long.parseLong(result)); + s.submitLock(K, 1L, () -> true, (result, error) -> token[0] = Long.parseLong(result)); // owner issues a token-carrying fetch (bypass) that stays in flight s.submitFetch(K, Long.toString(token[0]), fetchBody("BF"), () -> events.add("invalidLock:BF")); assertTrue(events.contains("fetch:BF")); @@ -345,7 +368,7 @@ public void lockForAlreadyDisconnectedOwnerIsNotGranted() { KeyedScheduler s = new KeyedScheduler(); final String[] outcome = {null}; // ownerStillConnected returns false: the client disconnected before the grant - s.submitLock(K, "gone", () -> false, + s.submitLock(K, 1L, () -> false, (result, error) -> outcome[0] = error != null ? "aborted" : "granted"); assertEquals("aborted", outcome[0]); assertEquals(0, s.getNumberOfLockedKeys()); @@ -379,7 +402,7 @@ public void unlockWithSentinelTokenZeroIsRejected() { @Test public void lockGrantReplyFailureReleasesLock() { KeyedScheduler s = new KeyedScheduler(); - s.submitLock(K, "owner", () -> true, (result, error) -> { + s.submitLock(K, 1L, () -> true, (result, error) -> { if (error == null) { throw new RuntimeException("reply delivery failed"); } @@ -439,7 +462,7 @@ public void doubleFiredExclusiveCompletionIsIdempotent() { @Test public void tokenIsValidWhenLockReplyFires() { KeyedScheduler s = new KeyedScheduler(); - s.submitLock(K, "owner", () -> true, (result, error) -> put(s, "OWNED", result)); + s.submitLock(K, 1L, () -> true, (result, error) -> put(s, "OWNED", result)); assertTrue(events.contains("bcast:OWNED")); assertFalse(events.contains("error:OWNED")); } @@ -448,7 +471,7 @@ public void tokenIsValidWhenLockReplyFires() { public void doubleFiredBypassCompletionIsIdempotent() { KeyedScheduler s = new KeyedScheduler(); final String[] token = new String[1]; - s.submitLock(K, "owner", () -> true, (result, error) -> token[0] = result); + s.submitLock(K, 1L, () -> true, (result, error) -> token[0] = result); put(s, "OWNED", token[0]); // bypasses the queue, broadcast pending Runnable owned = completions.get("OWNED"); From 1d8fc25664ea59375099d98691c310e7af33d44a Mon Sep 17 00:00:00 2001 From: diegosalvi Date: Wed, 15 Jul 2026 10:34:04 +0200 Subject: [PATCH 6/7] Make client-listener cleanup connection-identity aware on reconnect Mirror of the lock reconnect fix, on the near-cache holder registrations: - CacheServerEndpoint.connectionClosed now removes the client->connection mapping only if it still points to the closing connection (clientConnections.remove(id, con)), so a reconnected client's new connection is not unmapped by the late cleanup of the old, dead one. - CacheServer.clientDisconnected skips removeClientListeners when a newer connection has already taken over the same client id. Otherwise the dead connection's late cleanup would wipe the registrations the new connection just made, leaving it holding entries the server no longer tracks (stale, missed by future invalidations). When skipped, the dead connection's stale registrations are left as benign, self-healing phantoms instead of being wrongly wiped. --- .../java/blazingcache/server/CacheServer.java | 19 +++++++++++++++++-- .../server/CacheServerEndpoint.java | 5 ++++- 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java index 166e236..84ae2e1 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java @@ -589,8 +589,23 @@ private void executeOnHandler(String name, Runnable runnable) { } void clientDisconnected(String clientId, long connectionId) { - int count = cacheStatus.removeClientListeners(clientId); - LOGGER.log(Level.SEVERE, "client " + clientId + " (connection " + connectionId + ") disconnected, removed " + count + " key listeners"); + // Remove this client's key listeners only if no NEWER connection has already + // taken over the same client id (a reconnect). Otherwise the late cleanup of the + // dead connection would wipe the registrations the new connection just made, + // leaving it holding entries the server no longer knows about (stale). This + // mirrors the connection-identity handling of the application locks. When + // skipped, the dead connection's now-stale registrations are left as benign, + // self-healing phantoms for the new connection instead of being wrongly wiped. + CacheServerSideConnection current = acceptor.getActualConnectionFromClient(clientId); + boolean supersededByNewerConnection = current != null && current.getConnectionId() != connectionId; + if (supersededByNewerConnection) { + LOGGER.log(Level.SEVERE, "client " + clientId + " (connection " + connectionId + + ") disconnected, but connection " + current.getConnectionId() + + " already took over; leaving its key listeners in place"); + } else { + int count = cacheStatus.removeClientListeners(clientId); + LOGGER.log(Level.SEVERE, "client " + clientId + " (connection " + connectionId + ") disconnected, removed " + count + " key listeners"); + } // Release every application lock held or queued by THIS connection. Keying the // release on the connection id (not the client id) means a late cleanup of a dead // connection cannot release a lock that a new connection of the same client diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServerEndpoint.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServerEndpoint.java index 304e209..f8d1d99 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServerEndpoint.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServerEndpoint.java @@ -85,7 +85,10 @@ void connectionClosed(CacheServerSideConnection con) { LOGGER.log(Level.SEVERE, "connectionClosed {0}", con); connections.remove(con.getConnectionId()); if (con.getClientId() != null) { - clientConnections.remove(con.getClientId()); // to be remove only if the connection is the current connection + // remove the client mapping only if it still points to THIS connection: a + // newer connection of the same client (a reconnect) must not be unmapped by + // the late cleanup of the old, dead connection + clientConnections.remove(con.getClientId(), con); } } From 202fc1bcf5959b0519ffe761d61aa3c4e34d3c60 Mon Sep 17 00:00:00 2001 From: diegosalvi Date: Wed, 15 Jul 2026 19:30:47 +0200 Subject: [PATCH 7/7] invalidateByPrefix option B + reconnect/lock hardening from review - invalidateByPrefix: a single INVALIDATE_BY_PREFIX broadcast per client, with NO server-side holder removal. The previous two-phase per-key scheduler removal is reverted: it could self-deadlock behind a held (timeout-less) application lock on a matching key, early-ACK a coalesced concurrent invalidate, and wipe a freshly registered client. Not pruning means only the benign "server still thinks a client holds a key it dropped" divergence can occur (self-heals on the next invalidate/put/fetch, on expiry, or on disconnect); the dangerous "client holds / server forgot" direction is impossible. Removed CacheStatus.removeKey. - runLock: wrap the lock-owner liveness check in try/catch; a throw is treated as "owner not connected" and releases the just-published token instead of leaving the key locked forever and wedging the drain with draining=true. - listener cleanup made connection-identity aware and race-free: CacheStatus tracks the owning connection per client (connectionForClient), advanced FORWARD-ONLY via merge(Math::max) since connection ids are monotonic; removeClientListeners(clientId, connectionId) removes only if that connection still owns the registrations, atomically with a concurrent registerKeyForClient under the CacheStatus lock. connectionId is threaded through putEntry/loadEntry/fetchEntry. Closes the reconnect TOCTOU where a dead connection's late cleanup wiped a reconnected connection's fresh registrations. - fetchEntry: registerKeyForClient moved inside the completion guard so a doubly-fired reply cannot create a phantom holder. - removed a dead import; added CacheStatusReconnectTest and scheduler regression tests. --- .../java/blazingcache/server/CacheServer.java | 160 +++++++----------- .../server/CacheServerSideConnection.java | 6 +- .../java/blazingcache/server/CacheStatus.java | 65 ++++--- .../blazingcache/server/KeyedScheduler.java | 13 +- .../server/CacheStatusReconnectTest.java | 97 +++++++++++ .../server/KeyedSchedulerTest.java | 26 +++ 6 files changed, 227 insertions(+), 140 deletions(-) create mode 100644 blazingcache-core/src/test/java/blazingcache/server/CacheStatusReconnectTest.java diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java index 84ae2e1..909b440 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java @@ -30,7 +30,6 @@ import blazingcache.zookeeper.ZKClusterManager; import java.io.File; import java.util.ArrayList; -import java.util.Collections; import java.util.List; import java.util.Set; import java.util.concurrent.CountDownLatch; @@ -289,11 +288,11 @@ public void close() { channelsHandlers.shutdown(); } - public void putEntry(RawString key, byte[] data, long expiretime, String sourceClientId, String clientProvidedLockId, SimpleCallback onFinish) { + public void putEntry(RawString key, byte[] data, long expiretime, String sourceClientId, long sourceConnectionId, String clientProvidedLockId, SimpleCallback onFinish) { // Registering the source as a holder is the always-applied side effect (kept // even when the broadcast is coalesced away); the broadcast pushes the new // value to the other holders. - Runnable registration = () -> cacheStatus.registerKeyForClient(key, sourceClientId, expiretime); + Runnable registration = () -> cacheStatus.registerKeyForClient(key, sourceClientId, sourceConnectionId, expiretime); Consumer broadcast = (onComplete) -> { Set clientsForKey = cacheStatus.getClientsForKey(key); if (sourceClientId != null) { @@ -320,11 +319,11 @@ public void putEntry(RawString key, byte[] data, long expiretime, String sourceC scheduler.submitExclusive(key, KeyedScheduler.Verb.PUT, clientProvidedLockId, registration, broadcast, key, onFinish); } - public void loadEntry(RawString key, long expiretime, String sourceClientId, String clientProvidedLockId, SimpleCallback onFinish) { + public void loadEntry(RawString key, long expiretime, String sourceClientId, long sourceConnectionId, String clientProvidedLockId, SimpleCallback onFinish) { LOGGER.log(Level.FINEST, "loadEntry from {0}, key={1}", new Object[]{sourceClientId, key}); // load is local-only: it registers the source as a holder and does not // broadcast anything to the other clients. - Runnable registration = () -> cacheStatus.registerKeyForClient(key, sourceClientId, expiretime); + Runnable registration = () -> cacheStatus.registerKeyForClient(key, sourceClientId, sourceConnectionId, expiretime); scheduler.submitExclusive(key, KeyedScheduler.Verb.LOAD, clientProvidedLockId, registration, null, key, onFinish); } @@ -416,7 +415,7 @@ public void unregisterEntries(final List keys, String sourceClientId, } } - public void fetchEntry(RawString key, String clientId, String clientProvidedLockId, SimpleCallback onFinish) { + public void fetchEntry(RawString key, String clientId, long connectionId, String clientProvidedLockId, SimpleCallback onFinish) { // A fetch is a SHARED operation: concurrent fetches on the same key run in // parallel, while still being mutually exclusive with invalidate/put/load // (exclusive), preserving the ordering invariant that the fetch @@ -478,13 +477,21 @@ public void fetchEntry(RawString key, String clientId, String clientProvidedLock connection.sendFetchKeyMessage(remoteClientId, key, (result, error) -> { networkRequestsStatusMonitor.unregister(unicastRequestStatus); LOGGER.log(Level.FINE, "client " + remoteClientId + " answer to fetch :" + result, error); - if (result != null && result.type == Message.TYPE_ACK) { - // da questo momento consideriamo che il client abbia la entry in memoria - // anche se di fatto potrebbe succedere che il messaggio di risposta non arrivi mai - long expiretime = (long) result.parameters.get("expiretime"); - cacheStatus.registerKeyForClient(key, clientId, expiretime); + // Register the fetching client and reply exactly once: the reply + // callback can fire more than once (send failure then reply timeout), + // and the registration must not run on a late duplicate that arrives + // after an error already completed the fetch (it would create a + // phantom holder the client never actually cached). + if (finished.compareAndSet(false, true)) { + if (result != null && result.type == Message.TYPE_ACK) { + // da questo momento consideriamo che il client abbia la entry in memoria + // anche se di fatto potrebbe succedere che il messaggio di risposta non arrivi mai + long expiretime = (long) result.parameters.get("expiretime"); + cacheStatus.registerKeyForClient(key, clientId, connectionId, expiretime); + } + onComplete.run(); + onFinish.onResult(result, error); } - finish.onResult(result, error); }); } catch (Throwable t) { LOGGER.log(Level.SEVERE, "error in fetchEntry for key " + key + ", releasing the slot", t); @@ -497,85 +504,40 @@ public void fetchEntry(RawString key, String clientId, String clientProvidedLock public void invalidateByPrefix(RawString prefix, String sourceClientId, SimpleCallback onFinish) { Runnable action = () -> { - // Capture the clients to notify BEFORE phase 1: the per-key removals clear the - // holder registrations, so computing this set afterwards would miss the very - // clients that held the matching keys. - Set clientsToNotify = cacheStatus.getAllClientsWithListener(); - if (sourceClientId != null) { - clientsToNotify.remove(sourceClientId); - } - // Phase 2: notify every client with a SINGLE prefix broadcast (as before), so - // the message fan-out stays O(clients), not O(matching keys). Runs only after - // the server-side removals of phase 1 have completed, so that any PUT_ENTRY a - // concurrent put sent to a client is ordered before this invalidation on that - // client's channel. - Runnable broadcastPrefixToClients = () -> { - if (clientsToNotify.isEmpty()) { - onFinish.onResult(prefix, null); - return; - } - // holder registrations were already cleared per key in phase 1, so this - // broadcast only tells the clients to drop the matching keys locally - BroadcastRequestStatus invalidation = new BroadcastRequestStatus("invalidateByPrefix " + prefix + " from " + sourceClientId + " started at " + new java.sql.Timestamp(System.currentTimeMillis()), clientsToNotify, onFinish, null); - networkRequestsStatusMonitor.register(invalidation); - clientsToNotify.forEach((clientId) -> { - CacheServerSideConnection connection = acceptor.getActualConnectionFromClient(clientId); - if (connection == null) { - LOGGER.log(Level.SEVERE, "client " + clientId + " not connected, considering prefix " + prefix + " invalidated"); - invalidation.clientDone(clientId); - } else { - connection.sendPrefixInvalidationMessage(sourceClientId, prefix, invalidation); - } - }); - }; - - // Phase 1: clear each matching key on its own scheduler slot (no network), so - // the server-side removal is serialized with any concurrent put/load/fetch on - // that key. A single bulk prefix removal (the previous approach) did not go - // through the scheduler and could interleave with an in-flight putEntry on a - // matching key, leaving the client holding an entry the server no longer knows - // about (stale forever). - // - // Concurrency note - a put/load on a matching key that arrives AFTER its phase-1 - // removal but before the phase-2 broadcast re-registers that client, which then - // drops the key on the phase-2 broadcast: the server is left transiently - // believing the client still holds a key it dropped. This divergence is benign - // and self-healing (never the reverse "client holds it, server does not know" - // that would serve stale data): the next invalidate/put on the key, or the - // client disconnect, reconciles it, and a fetch routed to that client simply - // misses. The dangerous direction is prevented by the per-key serialization - // above plus running phase 2 only after phase 1, which orders any PUT_ENTRY a - // concurrent put sent to a client before this invalidation on its channel. - // Removing even the benign residual would require either a per-client bulk - // removal (reintroducing the race) or a per-key client broadcast (the O(keys) - // fan-out this design avoids), so it is accepted on purpose. + // Notify every client to drop its matching keys with a single prefix + // broadcast (fan-out is O(clients), and each client scans its own near-cache). // - // Behavioural change vs. the legacy prefix invalidation (which ignored locks - // and per-key coordination entirely): because phase 1 goes through the - // scheduler, a matching key that is currently held by an application lock - // (which has NO timeout) or has an invalidate/put broadcast in flight makes - // this prefix invalidation WAIT until that lock is released / that broadcast - // completes, before its clients are notified. - List keys = new ArrayList<>(); - for (RawString key : cacheStatus.getKeys()) { - if (key.startsWith(prefix)) { - keys.add(key); - } + // We deliberately do NOT prune the server-side holder registrations here. A + // bulk per-client removal (the legacy behaviour) races an in-flight putEntry + // on a matching key: the client can re-store the entry (from the put + // broadcast) after dropping it, while the server has already forgotten it + // holds the key, so future invalidations skip that client and it serves stale + // data. By not removing, the only possible divergence is the benign + // "server still thinks a client holds a key it dropped", which serves no stale + // data and self-heals on the next invalidate/put/fetch of the key, on expiry, + // or on the client's disconnect. Routing the removal through the per-key + // scheduler instead is not viable either: it would block the whole prefix + // invalidation behind any held application lock (which has no timeout) on a + // matching key, up to a self-deadlock when the lock owner is the caller. + Set clients = cacheStatus.getAllClientsWithListener(); + if (sourceClientId != null) { + clients.remove(sourceClientId); } - if (keys.isEmpty()) { - broadcastPrefixToClients.run(); + if (clients.isEmpty()) { + onFinish.onResult(prefix, null); return; } - AtomicInteger remaining = new AtomicInteger(keys.size()); - SimpleCallback perKeyRemoved = (result, error) -> { - if (remaining.decrementAndGet() == 0) { - broadcastPrefixToClients.run(); + BroadcastRequestStatus invalidation = new BroadcastRequestStatus("invalidateByPrefix " + prefix + " from " + sourceClientId + " started at " + new java.sql.Timestamp(System.currentTimeMillis()), clients, onFinish, null); + networkRequestsStatusMonitor.register(invalidation); + clients.forEach((clientId) -> { + CacheServerSideConnection connection = acceptor.getActualConnectionFromClient(clientId); + if (connection == null) { + LOGGER.log(Level.SEVERE, "client " + clientId + " not connected, considering prefix " + prefix + " invalidated"); + invalidation.clientDone(clientId); + } else { + connection.sendPrefixInvalidationMessage(sourceClientId, prefix, invalidation); } - }; - for (RawString key : keys) { - scheduler.submitExclusive(key, KeyedScheduler.Verb.INVALIDATE, null, - () -> cacheStatus.removeKey(key), null, key, perKeyRemoved); - } + }); }; executeOnHandler("invalidateByPrefix " + prefix, action); } @@ -589,21 +551,17 @@ private void executeOnHandler(String name, Runnable runnable) { } void clientDisconnected(String clientId, long connectionId) { - // Remove this client's key listeners only if no NEWER connection has already - // taken over the same client id (a reconnect). Otherwise the late cleanup of the - // dead connection would wipe the registrations the new connection just made, - // leaving it holding entries the server no longer knows about (stale). This - // mirrors the connection-identity handling of the application locks. When - // skipped, the dead connection's now-stale registrations are left as benign, - // self-healing phantoms for the new connection instead of being wrongly wiped. - CacheServerSideConnection current = acceptor.getActualConnectionFromClient(clientId); - boolean supersededByNewerConnection = current != null && current.getConnectionId() != connectionId; - if (supersededByNewerConnection) { - LOGGER.log(Level.SEVERE, "client " + clientId + " (connection " + connectionId - + ") disconnected, but connection " + current.getConnectionId() - + " already took over; leaving its key listeners in place"); - } else { - int count = cacheStatus.removeClientListeners(clientId); + // clientId is null for a connection that closed before completing the handshake + // (health checks, port scans, TLS failures): it never registered listeners nor + // acquired locks, so there is nothing keyed on it to clean up. Guarding here also + // avoids a ConcurrentHashMap.get(null) NPE in getActualConnectionFromClient. + if (clientId != null) { + // removeClientListeners removes this connection's registrations only if no + // newer connection of the same client has taken over in the meantime (a + // reconnect); the check is atomic with concurrent registerKeyForClient under + // the CacheStatus lock, so a reconnected connection's fresh registrations are + // never wiped by the late cleanup of the old, dead one. + int count = cacheStatus.removeClientListeners(clientId, connectionId); LOGGER.log(Level.SEVERE, "client " + clientId + " (connection " + connectionId + ") disconnected, removed " + count + " key listeners"); } // Release every application lock held or queued by THIS connection. Keying the diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java index bd157c0..e86b8ed 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java @@ -325,7 +325,7 @@ public void onResult(RawString result, Throwable error) { RawString _lockId = RawString.of(message.parameters.get("lockId")); String lockId = _lockId != null ? _lockId.toString() : null; server.addPendingOperations(1); - server.fetchEntry(key, clientId, lockId, new SimpleCallback() { + server.fetchEntry(key, clientId, connectionId, lockId, new SimpleCallback() { @Override public void onResult(Message result, Throwable error) { server.addPendingOperations(-1); @@ -385,7 +385,7 @@ public void onResult(RawString result, Throwable error) { RawString _lockId = RawString.of(message.parameters.get("lockId")); String lockId = _lockId != null ? _lockId.toString() : null; server.addPendingOperations(1); - server.putEntry(key, data, expiretime, clientId, lockId, new SimpleCallback() { + server.putEntry(key, data, expiretime, clientId, connectionId, lockId, new SimpleCallback() { @Override public void onResult(RawString result, Throwable error) { server.addPendingOperations(-1); @@ -405,7 +405,7 @@ public void onResult(RawString result, Throwable error) { RawString _lockId = RawString.of(message.parameters.get("lockId")); String lockId = _lockId != null ? _lockId.toString() : null; server.addPendingOperations(1); - server.loadEntry(key, expiretime, clientId, lockId, new SimpleCallback() { + server.loadEntry(key, expiretime, clientId, connectionId, lockId, new SimpleCallback() { @Override public void onResult(RawString result, Throwable error) { server.addPendingOperations(-1); diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java b/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java index 740225a..e61b03a 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java @@ -45,6 +45,11 @@ public class CacheStatus { private final Map> clientsForKey = new HashMap<>(); private final Map> keysForClient = new HashMap<>(); private final Map entryExpireTime = new HashMap<>(); + // clientId -> id of the connection that last registered a key for it. Used to make + // the disconnect cleanup connection-identity aware: a late cleanup of a dead + // connection must not wipe the registrations a new connection of the same client made + // after reconnecting. Guarded by {@link #lock}. + private final Map connectionForClient = new HashMap<>(); private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true); @Override @@ -57,8 +62,8 @@ public String toString() { } } - public void registerKeyForClient(RawString key, String client, long expiretime) { - LOGGER.log(Level.FINEST, "registerKeyForClient key={0} client={1}", new Object[]{key, client}); + public void registerKeyForClient(RawString key, String client, long connectionId, long expiretime) { + LOGGER.log(Level.FINEST, "registerKeyForClient key={0} client={1} connection={2}", new Object[]{key, client, connectionId}); lock.writeLock().lock(); try { Set clients = clientsForKey.get(key); @@ -74,6 +79,13 @@ public void registerKeyForClient(RawString key, String client, long expiretime) keysForClient.put(client, keys); } keys.add(key); + // record which connection owns this client's registrations now; a later + // disconnect of an older connection will not wipe them (see removeClientListeners). + // Advance the owner FORWARD-ONLY: connection ids are monotonic, so a late + // registration arriving from an older (already-superseded) connection - e.g. a + // delayed fetch reply, or a put/load that was queued behind a held lock - must + // not move ownership back to that dead connection and defeat the guard. + connectionForClient.merge(client, connectionId, Math::max); if (expiretime > 0) { entryExpireTime.put(key, expiretime); } else { @@ -148,6 +160,7 @@ public void removeKeyForClient(RawString key, String client) { keys.remove(key); if (keys.isEmpty()) { keysForClient.remove(client); + connectionForClient.remove(client); } } } finally { @@ -157,43 +170,26 @@ public void removeKeyForClient(RawString key, String client) { } /** - * Removes a key entirely, dropping the registration of every client that held it. - * Used by prefix invalidation, which clears each matching key on its own scheduler - * slot (serialized with concurrent per-key operations) before the clients are - * notified with a single prefix broadcast. - */ - void removeKey(RawString key) { - LOGGER.log(Level.FINEST, "removeKey key={0}", key); - lock.writeLock().lock(); - try { - Set clients = clientsForKey.remove(key); - entryExpireTime.remove(key); - if (clients != null) { - for (String client : clients) { - Set keys = keysForClient.get(client); - if (keys != null) { - keys.remove(key); - if (keys.isEmpty()) { - keysForClient.remove(client); - } - } - } - } - } finally { - lock.writeLock().unlock(); - } - } - - /** - * Removes all the key listeners of a disconnected client. + * Removes the key listeners of a disconnected connection, but ONLY if that connection + * is still the one that owns the client's registrations. If a newer connection of the + * same client (a reconnect) has registered in the meantime, this is a no-op so its + * fresh registrations are not wiped — the check and the removal happen atomically + * under the write lock, together with the concurrent registerKeyForClient, so there + * is no time-of-check/time-of-use gap. The dead connection's now-stale registrations + * are then left as benign, self-healing phantoms. * - * @return the number of listeners removed. Application locks held by the client are - * released separately by the {@link KeyedScheduler}, which owns the lock lifecycle. + * @return the number of listeners removed. Application locks are released separately + * by the {@link KeyedScheduler}, which owns the lock lifecycle. */ - int removeClientListeners(String clientId) { + int removeClientListeners(String clientId, long connectionId) { AtomicInteger count = new AtomicInteger(); lock.writeLock().lock(); try { + Long owner = connectionForClient.get(clientId); + if (owner != null && owner != connectionId) { + // a newer connection of this client already took over its registrations + return 0; + } Set keys = keysForClient.get(clientId); if (keys != null) { keys.forEach((key) -> { @@ -209,6 +205,7 @@ int removeClientListeners(String clientId) { }); } keysForClient.remove(clientId); + connectionForClient.remove(clientId); } finally { lock.writeLock().unlock(); } diff --git a/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java b/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java index b227969..63c0076 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java +++ b/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java @@ -561,8 +561,17 @@ private void runLock(KeySlot slot, Op op) { // Evaluate liveness outside the slot lock (it calls into the acceptor); the tiny // window between this check and the grant is backstopped by // releaseLocksForConnection, which runs after the connection is removed and finds - // the published heldToken. - boolean ownerConnected = op.ownerStillConnected == null || op.ownerStillConnected.getAsBoolean(); + // the published heldToken. Guard the call: heldToken is already published, so a + // throw escaping here would leave the key locked forever AND wedge drain() with + // draining left true. Treat any failure as "owner not connected" -> the lock is + // released below instead of leaked. + boolean ownerConnected; + try { + ownerConnected = op.ownerStillConnected == null || op.ownerStillConnected.getAsBoolean(); + } catch (Throwable t) { + LOGGER.log(Level.SEVERE, "error checking lock owner liveness for key " + slot.key + ", not granting the lock", t); + ownerConnected = false; + } boolean granted = false; slot.lock.lock(); try { diff --git a/blazingcache-core/src/test/java/blazingcache/server/CacheStatusReconnectTest.java b/blazingcache-core/src/test/java/blazingcache/server/CacheStatusReconnectTest.java new file mode 100644 index 0000000..319cd19 --- /dev/null +++ b/blazingcache-core/src/test/java/blazingcache/server/CacheStatusReconnectTest.java @@ -0,0 +1,97 @@ +/* + Licensed to Diennea S.r.l. under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. Diennea S.r.l. licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + + */ +package blazingcache.server; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import blazingcache.utils.RawString; +import org.junit.Test; + +/** + * The disconnect cleanup of a client's listeners must be tied to the CONNECTION that + * registered them, so a late cleanup of a dead connection cannot wipe the registrations a + * new connection of the same client made after reconnecting. + * + * @author diego.salvi + */ +public class CacheStatusReconnectTest { + + private static final RawString K = RawString.of("k"); + private static final RawString K2 = RawString.of("k2"); + + @Test + public void removeClientListenersIsConnectionIdentityAware() { + CacheStatus status = new CacheStatus(); + status.registerKeyForClient(K, "X", 1L, 0); + assertTrue(status.getKeysForClient("X").contains(K)); + + // a stale cleanup of a DIFFERENT connection id must not remove the registration + assertEquals(0, status.removeClientListeners("X", 99L)); + assertTrue(status.getKeysForClient("X").contains(K)); + + // the owning connection's cleanup removes it + assertEquals(1, status.removeClientListeners("X", 1L)); + assertTrue(status.getKeysForClient("X").isEmpty()); + } + + @Test + public void reconnectRegistrationSurvivesOldConnectionCleanup() { + CacheStatus status = new CacheStatus(); + // old connection 1 registered K + status.registerKeyForClient(K, "X", 1L, 0); + // client reconnects as connection 2 and registers K2 (ownership moves to 2) + status.registerKeyForClient(K2, "X", 2L, 0); + + // the late cleanup of the OLD connection 1 must be a no-op now + assertEquals(0, status.removeClientListeners("X", 1L)); + assertTrue("the reconnected connection's fresh registration must survive", + status.getKeysForClient("X").contains(K2)); + + // and the current connection's own cleanup still removes everything for the client + assertFalse(status.getKeysForClient("X").isEmpty()); + status.removeClientListeners("X", 2L); + assertTrue(status.getKeysForClient("X").isEmpty()); + } + + /** + * A late registration arriving from an OLDER (already-superseded) connection - e.g. a + * delayed fetch reply - must not move ownership back to that dead connection: the + * owner marker advances forward-only. Otherwise the reconnected connection's + * registrations would become wipeable by the old connection's cleanup. + */ + @Test + public void lateRegistrationFromOldConnectionDoesNotRegressOwnership() { + CacheStatus status = new CacheStatus(); + // the reconnected connection 2 owns the client's registrations + status.registerKeyForClient(K2, "X", 2L, 0); + // a delayed registration attributed to the OLD connection 1 lands afterwards + status.registerKeyForClient(K, "X", 1L, 0); + + // ownership must stay with the newer connection 2: the old connection's cleanup + // is a no-op and connection 2's live registration survives + assertEquals(0, status.removeClientListeners("X", 1L)); + assertTrue(status.getKeysForClient("X").contains(K2)); + + // connection 2's own cleanup removes everything (including the phantom K) + status.removeClientListeners("X", 2L); + assertTrue(status.getKeysForClient("X").isEmpty()); + } +} diff --git a/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java b/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java index 0c79e6b..15c469a 100644 --- a/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java +++ b/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java @@ -380,6 +380,32 @@ public void lockForAlreadyDisconnectedOwnerIsNotGranted() { assertTrue(s.getLockedKeys().isEmpty()); } + /** + * If the lock-owner liveness check itself throws, the lock (whose token is already + * published at dequeue) must not be left held and the key's drain must not be wedged: + * the failure is treated as "owner not connected", the lock is released, and queued + * work resumes. + */ + @Test + public void lockOwnerLivenessCheckThrowingReleasesAndDoesNotWedge() { + KeyedScheduler s = new KeyedScheduler(); + final boolean[] aborted = {false}; + s.submitLock(K, 1L, () -> { + throw new RuntimeException("liveness check boom"); + }, (result, error) -> { + if (error != null) { + aborted[0] = true; + } + }); + assertTrue("a throwing liveness check must abort the grant", aborted[0]); + assertEquals(0, s.getNumberOfLockedKeys()); + + put(s, "AFTER", null); + assertTrue("the key must not be wedged", events.contains("bcast:AFTER")); + fire("AFTER"); + assertTrue(s.getLockedKeys().isEmpty()); + } + /** * Unlocking with token 0 (the "no lock held" sentinel) on an existing-but-unlocked * slot must NOT produce a false success.