Skip to content

[MNG-8563] Improve the object cache memory usage #2091

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

Closed
gnodet wants to merge 1 commit into from
Closed

[MNG-8563] Improve the object cache memory usage #2091

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
186 changes: 159 additions & 27 deletions src/mdo/java/CacheManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,14 @@
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import org.apache.maven.api.annotations.ThreadSafe;

/**
Expand All @@ -36,9 +41,11 @@
@ThreadSafe
class CacheManager {
private static final CacheManager INSTANCE = new CacheManager();
private static final int BATCH_SIZE = 32;
private static final long CLEANUP_WAIT_MS = 100; // Time to wait when queue is empty

// Cache of object instances
private final ConcurrentHashMap<Class<?>, ConcurrentHashMap<Integer, List<CacheReference>>> instanceCache =
private final ConcurrentHashMap<Class<?>, WeakReference<ConcurrentHashMap<Integer, CacheNode>>> instanceCache =
new ConcurrentHashMap<>();

// Cache of reflected fields per class
Expand All @@ -47,22 +54,111 @@ class CacheManager {
// Reference queue
private final ReferenceQueue<Cacheable> refQueue = new ReferenceQueue<>();

/**
* Lightweight structure to handle hash collisions.
* Acts as both a WeakReference and a linked list node.
*/
private static class CacheNode extends WeakReference<Cacheable> {
final Class<?> clazz;
final int hash;
volatile CacheNode next; // null if this is the only element

CacheNode(Cacheable referent, ReferenceQueue<Cacheable> q, Class<?> clazz, int hash) {
super(referent, q);
this.clazz = clazz;
this.hash = hash;
}

/**
* Finds a matching object in this chain of nodes.
* Also cleans up any cleared references it encounters.
* Returns the node containing the match, or null if none found.
*/
CacheNode findMatch(Cacheable obj, ConcurrentHashMap<Integer, CacheNode> classCache) {
CacheNode prev = null;
CacheNode current = this;

while (current != null) {
Cacheable cached = current.get();
CacheNode next = current.next;

if (cached == null) {
// Remove cleared reference
if (prev == null) {
// This is the head node
if (next == null) {
// Last node in chain, remove entire entry
classCache.remove(hash);
} else {
// Replace head with next node
classCache.replace(hash, this, next);
}
} else {
// Skip this node
prev.next = next;
}
} else if (obj.cacheEquals(cached)) {
return current;
} else {
prev = current;
}
current = next;
}
return null;
}
}

private CacheManager() {
// Start a background thread to process cleared references
Thread cleanupThread = new Thread(() -> {
List<CacheNode> refBatch = new ArrayList<>(BATCH_SIZE);

while (!Thread.currentThread().isInterrupted()) {
try {
CacheReference ref = (CacheReference) refQueue.remove();
ConcurrentHashMap<Integer, List<CacheReference>> classCache = instanceCache.get(ref.clazz);
if (classCache != null) {
classCache.computeIfPresent(ref.hash, (k, list) -> {
synchronized (list) {
list.remove(ref);
return list.isEmpty() ? null : list;
// Try to collect a batch of references
refBatch.clear();
CacheNode ref;

// Poll for references until either batch is full or queue is empty
while (refBatch.size() < BATCH_SIZE &&
(ref = (CacheNode)refQueue.poll()) != null) {
refBatch.add(ref);
}

if (refBatch.isEmpty()) {
// Queue is empty, wait for new references
ref = (CacheNode)refQueue.remove(CLEANUP_WAIT_MS);
if (ref != null) {
refBatch.add(ref);
}
}

// Process the batch
if (!refBatch.isEmpty()) {
// Group references by class for more efficient processing
Map<Class<?>, List<CacheNode>> byClass = refBatch.stream()
.collect(Collectors.groupingBy(node -> node.clazz));

for (var entry : byClass.entrySet()) {
WeakReference<ConcurrentHashMap<Integer, CacheNode>> weakClassCache =
instanceCache.get(entry.getKey());
if (weakClassCache != null) {
ConcurrentHashMap<Integer, CacheNode> classCache = weakClassCache.get();
if (classCache != null) {
// Process all references for this class
for (CacheNode node : entry.getValue()) {
CacheNode head = classCache.get(node.hash);
if (head != null) {
head.findMatch(null, classCache);
}
}

// If the class cache is empty, remove its weak reference
if (classCache.isEmpty()) {
instanceCache.remove(entry.getKey(), weakClassCache);
}
}
}
});
if (classCache.isEmpty()) {
instanceCache.remove(ref.clazz);
}
}
} catch (InterruptedException e) {
Expand Down Expand Up @@ -207,31 +303,67 @@ public <T extends Cacheable> T cached(T obj) {
}

Class<?> clazz = obj.getClass();
ConcurrentHashMap<Integer, List<CacheReference>> classCache =
instanceCache.computeIfAbsent(clazz, k -> new ConcurrentHashMap<>());

ConcurrentHashMap<Integer, CacheNode> classCache = getOrCreateClassCache(clazz);
int cacheHash = obj.cacheIdentityHash();

List<CacheReference> refs = classCache.compute(cacheHash, (k, oldList) -> {
if (oldList == null) {
oldList = new ArrayList<>();
while (true) {
CacheNode head = classCache.get(cacheHash);
if (head == null) {
// No existing entry, try to add new one
CacheNode newNode = new CacheNode(obj, refQueue, clazz, cacheHash);
if (classCache.putIfAbsent(cacheHash, newNode) == null) {
return obj; // Successfully added
}
continue; // Race condition, try again
}
List<CacheReference> newList = new ArrayList<>(oldList);
newList.removeIf(ref -> ref == null || ref.get() == null);
return newList;
});

synchronized (refs) {
for (WeakReference<Cacheable> ref : refs) {
Cacheable cached = ref.get();
if (cached != null && obj.cacheEquals(cached)) {
// Look for existing match
CacheNode match = head.findMatch(obj, classCache);
if (match != null) {
Cacheable cached = match.get();
if (cached != null) {
return (T) cached;
}
continue; // Reference was cleared, try again
}

// No match found, try to add to chain
synchronized (head) { // Synchronize on head node for chain modifications
// Verify head is still valid
if (classCache.get(cacheHash) != head) {
continue; // Head changed, try again
}

// Add new node to chain
CacheNode newNode = new CacheNode(obj, refQueue, clazz, cacheHash);
newNode.next = head.next;
head.next = newNode;
return obj;
}
}
}

private ConcurrentHashMap<Integer, CacheNode> getOrCreateClassCache(Class<?> clazz) {
WeakReference<ConcurrentHashMap<Integer, CacheNode>> weakRef = instanceCache.get(clazz);
ConcurrentHashMap<Integer, CacheNode> classCache = (weakRef != null) ? weakRef.get() : null;

refs.add(new CacheReference(obj, refQueue, clazz, cacheHash));
return obj;
if (classCache == null) {
classCache = new ConcurrentHashMap<>();
weakRef = new WeakReference<>(classCache);
WeakReference<ConcurrentHashMap<Integer, CacheNode>> existing =
instanceCache.putIfAbsent(clazz, weakRef);

if (existing != null) {
ConcurrentHashMap<Integer, CacheNode> existingCache = existing.get();
if (existingCache != null) {
classCache = existingCache;
} else {
instanceCache.replace(clazz, existing, weakRef);
}
}
}

return classCache;
}

/**
Expand Down
Loading