diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java index 05c7db0..909b440 100644 --- a/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java +++ b/blazingcache-core/src/main/java/blazingcache/server/CacheServer.java @@ -30,9 +30,7 @@ import blazingcache.zookeeper.ZKClusterManager; import java.io.File; 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 +41,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 +59,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; @@ -290,154 +288,61 @@ public void close() { channelsHandlers.shutdown(); } - 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)); - 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, 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); + 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, sourceConnectionId, expiretime); + Consumer broadcast = (onComplete) -> { + Set clientsForKey = cacheStatus.getClientsForKey(key); + if (sourceClientId != null) { + clientsForKey.remove(sourceClientId); } - }; - executeOnHandler("putEntry " + sourceClientId + "," + key, action); - } - - 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)); + LOGGER.log(Level.FINEST, "putEntry from {0}, key={1}, clientsForKey:{2}", new Object[]{sourceClientId, key, clientsForKey}); + if (clientsForKey.isEmpty()) { + onComplete.run(); 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); + 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); } - }; - 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); + scheduler.submitExclusive(key, KeyedScheduler.Verb.PUT, clientProvidedLockId, registration, broadcast, 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); + 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, sourceConnectionId, expiretime); + scheduler.submitExclusive(key, KeyedScheduler.Verb.LOAD, clientProvidedLockId, registration, null, key, onFinish); } - 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); + public void invalidateKey(RawString key, String sourceClientId, String clientProvidedLockId, SimpleCallback onFinish) { + // 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); @@ -466,110 +371,105 @@ 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); + public void lockKey(RawString key, String sourceClientId, CacheServerSideConnection connection, SimpleCallback onFinish) { + // 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); } 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) -> { + 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 + // 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); @@ -577,33 +477,48 @@ 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); } - finishAndReleaseLock.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) { Runnable action = () -> { - // LOCKS ?? + // 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). + // + // 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); @@ -612,10 +527,7 @@ public void invalidateByPrefix(RawString prefix, String sourceClientId, SimpleCa 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); - }); - + 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); @@ -638,19 +550,25 @@ 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); - }); - - }); + void clientDisconnected(String clientId, long connectionId) { + // 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 + // 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() { @@ -681,12 +599,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/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); } } diff --git a/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java b/blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java index c82aedc..e86b8ed 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); @@ -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); @@ -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/CacheStatus.java b/blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java index 01f50a4..e61b03a 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; @@ -46,9 +45,12 @@ 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); - private final Map>> remoteLocks = new HashMap<>(); - private final ReentrantLock remoteLocksLock = new ReentrantLock(true); @Override public String toString() { @@ -60,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); @@ -77,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 { @@ -151,6 +160,7 @@ public void removeKeyForClient(RawString key, String client) { keys.remove(key); if (keys.isEmpty()) { keysForClient.remove(client); + connectionForClient.remove(client); } } } finally { @@ -159,61 +169,27 @@ 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(); - } - } - - 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 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 are released separately + * by the {@link KeyedScheduler}, which owns the lock lifecycle. + */ + 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) -> { @@ -229,17 +205,11 @@ ClientRemovalResult removeClientListeners(String clientId) { }); } keysForClient.remove(clientId); + connectionForClient.remove(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 +247,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..63c0076 --- /dev/null +++ b/blazingcache-core/src/main/java/blazingcache/server/KeyedScheduler.java @@ -0,0 +1,841 @@ +/* + 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; + op.resultKey = key; // only used for diagnostics (e.g. error logging) + 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 #releaseLocksForConnection} on disconnect) releases it. Operations + * carrying the matching token bypass the queue in the meantime. + *

+ * 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, long ownerConnectionId, + BooleanSupplier ownerStillConnected, SimpleCallback onFinish) { + long token = tokenGenerator.incrementAndGet(); + Op op = new Op(Verb.LOCK, Mode.EXCLUSIVE); + op.ownerConnectionId = ownerConnectionId; + 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.lockOwnerConnectionId = 0; + 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 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 releaseLocksForConnection(long connectionId) { + slots.forEach((key, slot) -> { + List> cancelledLocks = null; + boolean changed = false; + slot.lock.lock(); + try { + if (slot.heldToken != 0 && slot.lockOwnerConnectionId == connectionId) { + slot.heldToken = 0; + slot.lockOwnerConnectionId = 0; + changed = true; + } + for (Iterator it = slot.queue.iterator(); it.hasNext();) { + Op queued = it.next(); + if (queued.verb == Verb.LOCK && queued.ownerConnectionId == connectionId) { + 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("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); + } + } + } + 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.lockOwnerConnectionId = head.ownerConnectionId; + } 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 + 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. 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 { + if (slot.heldToken == op.lockToken) { + if (ownerConnected) { + granted = true; + } else { + // release the lock we just published: the owner disconnected + slot.heldToken = 0; + slot.lockOwnerConnectionId = 0; + } + } + } 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 { + // 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.lockOwnerConnectionId = 0; + } + } 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 long lockOwnerConnectionId; + 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 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(); + + // 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/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/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/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/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..15c469a --- /dev/null +++ b/blazingcache-core/src/test/java/blazingcache/server/KeyedSchedulerTest.java @@ -0,0 +1,539 @@ +/* + 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, 1L, () -> 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 releaseLocksForConnectionReleasesHeldLockAndResumesDrain() { + KeyedScheduler s = new KeyedScheduler(); + s.submitLock(K, 1L, () -> true, (result, error) -> { + }); + put(s, "WAIT", null); // queued behind the lock + assertFalse(events.contains("bcast:WAIT")); + + 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()); + } + + /** + * 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 releaseLocksForConnectionCancelsQueuedLock() { + KeyedScheduler s = new KeyedScheduler(); + final long[] owner1Token = new long[1]; + s.submitLock(K, 1L, () -> true, + (result, error) -> owner1Token[0] = Long.parseLong(result)); + + final boolean[] cancelled = {false}; + s.submitLock(K, 2L, () -> true, + (result, error) -> { + if (error != null) { + cancelled[0] = true; + } else { + events.add("granted2"); + } + }); + + 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) + 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, 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")); + + 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, 1L, () -> 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()); + } + + /** + * 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. + */ + @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, 1L, () -> 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, 1L, () -> 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, 1L, () -> 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()); + } +}