Skip to content

Make user cache flushing race-safe #297

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

Merged
BenCodez merged 6 commits into from
Aug 25, 2026
Merged

Make user cache flushing race-safe #297

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
fd1676d
Make user cache change flushing race-safe
BenCodez Aug 24, 2026
3ddabdb
Test user cache flush scheduling
BenCodez Aug 24, 2026
439729d
Preserve user changes when flush preparation fails
BenCodez Aug 25, 2026
fe9d4dd
Test failed user cache flush restoration
BenCodez Aug 25, 2026
618cb02
Wait for in-flight user cache flushes before dump
BenCodez Aug 25, 2026
fbc7a52
Test dump waits for in-flight cache batches
BenCodez Aug 25, 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
189 changes: 124 additions & 65 deletions AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/usercache/UserDataCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;

import com.bencodez.advancedcore.api.user.AdvancedCoreUser;
Expand All @@ -22,8 +23,9 @@ public class UserDataCache {

private Queue<UserDataChange> cachedChanges;

private UserDataManager manager;
private final UserDataManager manager;
private boolean scheduled = false;
private int inFlightBatches = 0;
@Getter
private UUID uuid;

Expand All @@ -35,18 +37,20 @@ public UserDataCache(UserDataManager manager, UUID uuid) {
}

public synchronized void addChange(UserDataChange change, boolean queue) {
if (change == null || cache == null || cachedChanges == null) {
return;
}
cache.put(change.getKey(), change.toUserDataValue());
if (queue) {
cachedChanges.add(change);
if (!scheduled) {
scheduleChanges();
}
}

}

public UserDataCache cache() {
if (uuid != null) {
public synchronized UserDataCache cache() {
if (uuid != null && cache != null) {
AdvancedCoreUser user = getUser();
ArrayList<String> keys = user.getUserData().getKeys();
HashMap<String, DataValue> data = user.getUserData().getValues();
Expand All @@ -58,22 +62,18 @@ public UserDataCache cache() {
DataValue dataValue = data.get(key);
manager.getPlugin().devDebug("Caching " + dataValue.getTypeName() + " " + key + " for "
+ uuid.toString() + ", value: " + dataValue.toString());
// temp try/catch to prevent plugin failures
try {
if (cache.containsKey(key)) {
if (!cache.get(key).toString().equals(dataValue.toString())) {
changedKeys.add(key);
}
if (cache.containsKey(key) && !cache.get(key).toString().equals(dataValue.toString())) {
changedKeys.add(key);
}
} catch (Exception e) {
e.printStackTrace();
manager.getPlugin().debug(e);
}
cache.put(key, dataValue);
} else {
manager.getPlugin().devDebug("Loading default cache value for " + key + " for " + uuid.toString());
cache.put(key, dataKey.getDefault());
}

}
if (!changedKeys.isEmpty()) {
manager.getPlugin().getUserManager().onChange(user, ArrayUtils.convert(changedKeys));
Expand All @@ -85,14 +85,16 @@ public UserDataCache cache() {
return this;
}

public void clearCache() {
public synchronized void clearCache() {
if (hasChangesToProcess()) {
processChanges();
}
cache.clear();
if (cache != null) {
cache.clear();
}
}

public void clearChanges() {
public synchronized void clearChanges() {
if (hasChangesToProcess()) {
processChanges();
}
Expand All @@ -102,9 +104,12 @@ public void displayCache() {
manager.getPlugin().devDebug(displayCacheStringList().toString());
}

public ArrayList<String> displayCacheStringList() {
public synchronized ArrayList<String> displayCacheStringList() {
ArrayList<String> list = new ArrayList<>();
list.add("Current cache for " + uuid + ": ");
if (cache == null) {
return list;
}
for (Entry<String, DataValue> entry : getCache().entrySet()) {
if (entry.getValue().isBoolean()) {
list.add(entry.getKey() + "=" + entry.getValue().getBoolean());
Expand All @@ -118,87 +123,141 @@ public ArrayList<String> displayCacheStringList() {
}

public void dump() {
if (hasChangesToProcess()) {
while (true) {
processChanges();
synchronized (this) {
while (inFlightBatches > 0) {
try {
wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
if (cachedChanges != null && !cachedChanges.isEmpty()) {
continue;
}
cache = null;
cachedChanges = null;
uuid = null;
scheduled = false;
return;
}
}
cache = null;
cachedChanges = null;
uuid = null;
}

public AdvancedCoreUser getUser() {
return manager.getPlugin().getUserManager().getUser(uuid, false);
}

public boolean hasCache() {
return !cache.isEmpty();
public synchronized boolean hasCache() {
return cache != null && !cache.isEmpty();
}

public boolean hasChangesToProcess() {
return !cachedChanges.isEmpty();
public synchronized boolean hasChangesToProcess() {
return cachedChanges != null && !cachedChanges.isEmpty();
}

public boolean isCached(String key) {
if (cache != null) {
return cache.containsKey(key);
}
return false;
public synchronized boolean isCached(String key) {
return cache != null && cache.containsKey(key);
}

public void processChanges() {
if (uuid != null) {
if (cachedChanges.size() > 0) {
manager.getPlugin()
.extraDebug("Processing changes for " + uuid.toString() + ", Changes: " + cachedChanges.size());
AdvancedCoreUser user = getUser();
HashMap<String, DataValue> values = new HashMap<>();
ArrayList<String> keys = new ArrayList<>();
while (!cachedChanges.isEmpty()) {
UserDataChange change = cachedChanges.poll();
values.put(change.getKey(), change.toUserDataValue());
keys.add(change.getKey());
change.dump();
// manager.getPlugin().extraDebug("Processing change for " + change.getKey());
}
if (!values.isEmpty()) {
user.getUserData().setValues(values);
}
manager.getPlugin().getUserManager().onChange(user, ArrayUtils.convert(keys));
UUID currentUuid;
ArrayList<UserDataChange> changes = new ArrayList<>();
synchronized (this) {
currentUuid = uuid;
if (currentUuid == null || cachedChanges == null || cachedChanges.isEmpty()) {
return;
}
UserDataChange change;
while ((change = cachedChanges.poll()) != null) {
changes.add(change);
Comment thread
BenCodez marked this conversation as resolved.
Resolved
}
inFlightBatches++;
}

boolean persisted = false;
try {
manager.getPlugin().extraDebug("Processing changes for " + currentUuid + ", Changes: " + changes.size());
AdvancedCoreUser user = manager.getPlugin().getUserManager().getUser(currentUuid, false);
HashMap<String, DataValue> values = new HashMap<>();
ArrayList<String> keys = new ArrayList<>();
for (UserDataChange change : changes) {
values.put(change.getKey(), change.toUserDataValue());
keys.add(change.getKey());
}
if (!values.isEmpty()) {
user.getUserData().setValues(values);
}
persisted = true;
manager.getPlugin().getUserManager().onChange(user, ArrayUtils.convert(keys));
for (UserDataChange change : changes) {
change.dump();
}
} catch (RuntimeException | Error e) {
if (!persisted) {
requeueChanges(changes);
}
throw e;
} finally {
finishInFlightBatch();
}
}

public void processChangesAsync() {
if (uuid != null && cachedChanges.size() > 0) {
manager.getPlugin().getTimer().execute(new Runnable() {
private synchronized void requeueChanges(ArrayList<UserDataChange> changes) {
if (changes == null || changes.isEmpty() || cachedChanges == null) {
return;
}
Queue<UserDataChange> restored = new ConcurrentLinkedQueue<>();
restored.addAll(changes);
restored.addAll(cachedChanges);
cachedChanges = restored;
}

@Override
public void run() {
processChanges();
}
});
private synchronized void finishInFlightBatch() {
if (inFlightBatches > 0) {
inFlightBatches--;
}
notifyAll();
}

public void processChangesAsync() {
if (uuid != null && hasChangesToProcess()) {
manager.getPlugin().getTimer().execute(this::processChanges);
}
}

private void scheduleChanges() {
private synchronized void scheduleChanges() {
if (scheduled || cachedChanges == null || cachedChanges.isEmpty()) {
return;
}
manager.getPlugin().debug("Schedule changes");
scheduled = true;

manager.getTimer().schedule(new Runnable() {

@Override
public void run() {
try {
manager.getTimer().schedule(() -> {
try {
processChanges();
scheduled = false;
} catch (Exception e) {
manager.getPlugin().debug(e);
} finally {
onScheduledFlushComplete();
}
}
}, 3, TimeUnit.SECONDS);
}, 3, TimeUnit.SECONDS);
} catch (RejectedExecutionException e) {
scheduled = false;
manager.getPlugin().debug(e);
}
}

private synchronized void onScheduledFlushComplete() {
scheduled = false;
if (cachedChanges != null && !cachedChanges.isEmpty()) {
scheduleChanges();
}
}

public void updateCache(HashMap<String, DataValue> tempCache) {
cache = tempCache;
public synchronized void updateCache(HashMap<String, DataValue> tempCache) {
cache = tempCache == null ? new HashMap<>() : new HashMap<>(tempCache);
}
}
Loading
Loading