Skip to content

Rework server-side per-key coordination with a non-blocking scheduler #206

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
diegosalvi wants to merge 7 commits into
base: master
Choose a base branch
from
Open

Rework server-side per-key coordination with a non-blocking scheduler #206

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
2430f92
Rework server-side per-key coordination with a non-blocking scheduler
diegosalvi Jul 7, 2026
e78da97
Address review: lock connection-identity check and per-key invalidate…
diegosalvi Jul 8, 2026
31f9bb9
invalidateByPrefix: coalesce client notification into a single broadcast
diegosalvi Jul 8, 2026
3b2bcb7
Document the benign put/load-vs-invalidateByPrefix race between the t…
diegosalvi Jul 8, 2026
dd1fca8
Address second review round
diegosalvi Jul 15, 2026
1d8fc25
Make client-listener cleanup connection-identity aware on reconnect
diegosalvi Jul 15, 2026
202fc1b
invalidateByPrefix option B + reconnect/lock hardening from review
diegosalvi Jul 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
430 changes: 174 additions & 256 deletions blazingcache-core/src/main/java/blazingcache/server/CacheServer.java
View file
Open in desktop

Large diffs are not rendered by default.

5 changes: 4 additions & 1 deletion blazingcache-core/src/main/java/blazingcache/server/CacheServerEndpoint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}

Expand Down
10 changes: 5 additions & 5 deletions blazingcache-core/src/main/java/blazingcache/server/CacheServerSideConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String>() {
server.lockKey(key, clientId, this, new SimpleCallback<String>() {
@Override
public void onResult(String result, Throwable error) {
server.addPendingOperations(-1);
Expand Down Expand Up @@ -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<Message>() {
server.fetchEntry(key, clientId, connectionId, lockId, new SimpleCallback<Message>() {
@Override
public void onResult(Message result, Throwable error) {
server.addPendingOperations(-1);
Expand Down Expand Up @@ -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<RawString>() {
server.putEntry(key, data, expiretime, clientId, connectionId, lockId, new SimpleCallback<RawString>() {
@Override
public void onResult(RawString result, Throwable error) {
server.addPendingOperations(-1);
Expand All @@ -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<RawString>() {
server.loadEntry(key, expiretime, clientId, connectionId, lockId, new SimpleCallback<RawString>() {
@Override
public void onResult(RawString result, Throwable error) {
server.addPendingOperations(-1);
Expand Down Expand Up @@ -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
Expand Down
142 changes: 35 additions & 107 deletions blazingcache-core/src/main/java/blazingcache/server/CacheStatus.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -46,9 +45,12 @@ public class CacheStatus {
private final Map<RawString, Set<String>> clientsForKey = new HashMap<>();
private final Map<String, Set<RawString>> keysForClient = new HashMap<>();
private final Map<RawString, Long> 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<String, Long> connectionForClient = new HashMap<>();
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
private final Map<String, Map<RawString, List<LockID>>> remoteLocks = new HashMap<>();
private final ReentrantLock remoteLocksLock = new ReentrantLock(true);

@Override
public String toString() {
Expand All @@ -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<String> clients = clientsForKey.get(key);
Expand All @@ -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 {
Expand Down Expand Up @@ -151,6 +160,7 @@ public void removeKeyForClient(RawString key, String client) {
keys.remove(key);
if (keys.isEmpty()) {
keysForClient.remove(client);
connectionForClient.remove(client);
}
}
} finally {
Expand All @@ -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<RawString> keys = keysForClient.get(client);
Set<RawString> 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<String> 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<RawString, List<LockID>> locks;

public ClientRemovalResult(int listenersCount, Map<RawString, List<LockID>> locks) {
this.listenersCount = listenersCount;
this.locks = locks;
}

public int getListenersCount() {
return listenersCount;
}

public Map<RawString, List<LockID>> 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<RawString> keys = keysForClient.get(clientId);
if (keys != null) {
keys.forEach((key) -> {
Expand All @@ -229,17 +205,11 @@ ClientRemovalResult removeClientListeners(String clientId) {
});
}
keysForClient.remove(clientId);
connectionForClient.remove(clientId);
} finally {
lock.writeLock().unlock();
}
Map<RawString, List<LockID>> locksForClient;
remoteLocksLock.lock();
try {
locksForClient = remoteLocks.remove(clientId);
} finally {
remoteLocksLock.unlock();
}
return new ClientRemovalResult(count.get(), locksForClient);
return count.get();
}

Set<String> getAllClientsWithListener() {
Expand Down Expand Up @@ -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<RawString, List<LockID>> locksForClient = remoteLocks.get(sourceClientId);
if (locksForClient == null) {
locksForClient = new HashMap<>();
remoteLocks.put(sourceClientId, locksForClient);
}
List<LockID> 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<RawString, List<LockID>> locksForClient = remoteLocks.get(sourceClientId);
if (locksForClient == null) {
return;
}
List<LockID> 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();
}
}
}
Loading
Loading