From 784cf4d09f31c2d2072a64813a22bf14c532e9d2 Mon Sep 17 00:00:00 2001 From: Rishabh Date: Fri, 6 Feb 2026 00:20:24 +0530 Subject: [PATCH 1/3] active ttl using delayqueue --- src/main/java/ExpiryEntry.java | 34 +++++++ src/main/java/Jmap.java | 41 ++++++-- src/main/java/KVStore.java | 37 ++++--- src/main/java/TTLManager.java | 59 ++++++++++++ src/test/java/TTLTest.java | 171 +++++++++++++++++++++++++++++++++ 5 files changed, 322 insertions(+), 20 deletions(-) create mode 100644 src/main/java/ExpiryEntry.java create mode 100644 src/main/java/TTLManager.java create mode 100644 src/test/java/TTLTest.java diff --git a/src/main/java/ExpiryEntry.java b/src/main/java/ExpiryEntry.java new file mode 100644 index 0000000..b3fafcf --- /dev/null +++ b/src/main/java/ExpiryEntry.java @@ -0,0 +1,34 @@ +import java.util.concurrent.Delayed; +import java.util.concurrent.TimeUnit; + +public class ExpiryEntry implements Delayed { + private final String key; + private final long expiryTime; // absolute time in millis + + public ExpiryEntry(String key, long expiryTime) { + this.key = key; + this.expiryTime = expiryTime; + } + + public String getKey() { + return key; + } + + public long getExpiryTime() { + return expiryTime; + } + + @Override + public long getDelay(TimeUnit unit) { + long diff = expiryTime - System.currentTimeMillis(); + return unit.convert(diff, TimeUnit.MILLISECONDS); + } + + @Override + public int compareTo(Delayed other) { + if (other instanceof ExpiryEntry) { + return Long.compare(this.expiryTime, ((ExpiryEntry) other).expiryTime); + } + return Long.compare(this.getDelay(TimeUnit.MILLISECONDS), other.getDelay(TimeUnit.MILLISECONDS)); + } +} diff --git a/src/main/java/Jmap.java b/src/main/java/Jmap.java index 96724b5..2d2e427 100644 --- a/src/main/java/Jmap.java +++ b/src/main/java/Jmap.java @@ -1,4 +1,3 @@ -import java.security.Timestamp; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; @@ -40,7 +39,7 @@ public Jmap(float threshold) { private int hashFunction(K key, int mapCapacity) { // so when a key is provided it returns a hash value; int h = Math.abs(key.hashCode()); - //same as concurrentHahsMap; + // same as concurrentHahsMap; h += (h << 15) ^ 0xffffcd7d; h ^= (h >>> 10); h += (h << 3); @@ -51,7 +50,7 @@ private int hashFunction(K key, int mapCapacity) { // how do i efficiently put a value with ttl? // do i overload the method but that would result in duplicate code; - //do i put a condition + // do i put a condition public void put(K key, V value) { readLock.lock(); @@ -62,7 +61,8 @@ public void put(K key, V value) { boolean newNodePlaced = false; Jnode node = new Jnode<>(tableIndex, key, value); - // System.out.println(Thread.currentThread().getName() + " got lock " + tableIndex); + // System.out.println(Thread.currentThread().getName() + " got lock " + + // tableIndex); if (table[tableIndex] == null) { table[tableIndex] = node; node_ct.getAndIncrement(); @@ -110,7 +110,7 @@ public void put(K key, V value) { } } - //put/putexp on an existing key overwrites the ttl of the prev key; + // put/putexp on an existing key overwrites the ttl of the prev key; public void putexp(K key, V value, long ttl) { readLock.lock(); int tableIndex = hashFunction(key, size.get()); @@ -120,7 +120,8 @@ public void putexp(K key, V value, long ttl) { boolean newNodePlaced = false; Jnode node = new Jnode<>(tableIndex, key, value, ttl); - // System.out.println(Thread.currentThread().getName() + " got lock " + tableIndex); + // System.out.println(Thread.currentThread().getName() + " got lock " + + // tableIndex); if (table[tableIndex] == null) { table[tableIndex] = node; node_ct.getAndIncrement(); @@ -217,7 +218,8 @@ public V get(K key) { if (value.key.equals(key) && (value.ttl == -1 || System.currentTimeMillis() <= value.ttl)) { return value.value; } else if (value.key.equals(key) && value.ttl != -1 && System.currentTimeMillis() > value.ttl) { - //will have to manually remove the entry, can't use remove() bcoz it'll deadlock; + // will have to manually remove the entry, can't use remove() bcoz it'll + // deadlock; if (prev == null) { table[tableIndex] = value.next; } else { @@ -238,6 +240,31 @@ public V get(K key) { } } + /** + * Returns the expiry time (absolute millis) for a key, or null if key doesn't + * exist or has no TTL. + * Used by TTLManager to validate stale entries before deletion. + */ + public Long getExpiry(K key) { + readLock.lock(); + int tableIndex = hashFunction(key, size.get()); + ReadWriteLock bucketLock = rwLocks[tableIndex]; + bucketLock.readLock().lock(); + try { + Jnode node = table[tableIndex]; + while (node != null) { + if (node.key.equals(key)) { + return node.ttl == -1 ? null : node.ttl; + } + node = node.next; + } + return null; + } finally { + bucketLock.readLock().unlock(); + readLock.unlock(); + } + } + public void resize_put(K key, V value, Jnode[] newTable) { int tableIndex = hashFunction(key, newTable.length); Jnode node = new Jnode<>(tableIndex, key, value); diff --git a/src/main/java/KVStore.java b/src/main/java/KVStore.java index 6a5cee3..3f8e967 100644 --- a/src/main/java/KVStore.java +++ b/src/main/java/KVStore.java @@ -14,14 +14,17 @@ public class KVStore { - // private static ConcurrentHashMap map = new ConcurrentHashMap<>(16, 0.75f); + // private static ConcurrentHashMap map = new + // ConcurrentHashMap<>(16, 0.75f); private static Jmap map = new Jmap<>(0.75f); + private static TTLManager ttlManager; private static PrintWriter logWriter; private static final String LOG_FILE = "src/main/logs/logs.txt"; public static void main(String[] args) throws InterruptedException { - // initializeLog(); - // loadLog(); + ttlManager = new TTLManager(map); + initializeLog(); + loadLog(); interactiveMode(); // stressTest(); } @@ -40,7 +43,8 @@ private static void initializeLog() { private static void loadLog() { File file = new File(LOG_FILE); - if (!file.exists()) return; + if (!file.exists()) + return; System.out.println("Loading data from log file..."); try (BufferedReader reader = new BufferedReader(new FileReader(file))) { @@ -58,7 +62,8 @@ private static void loadLog() { try { map.remove(key); count++; - } catch (Exception ignored) {} + } catch (Exception ignored) { + } } } } @@ -79,7 +84,7 @@ private static void log(String op, String key, String value) { } } - //to deal with multi word keys/values + // to deal with multi word keys/values private static String[] parseInput(String input) { ArrayList parts = new ArrayList<>(); StringBuilder current = new StringBuilder(); @@ -114,7 +119,8 @@ public static void interactiveMode() { while (true) { System.out.print("> "); String input = scanner.nextLine().trim(); - if (input.isEmpty()) continue; + if (input.isEmpty()) + continue; String[] parts = parseInput(input); @@ -133,11 +139,14 @@ public static void interactiveMode() { break; case "putexp": if (parts.length == 4) { - map.putexp(parts[1], parts[2], Long.parseLong(parts[3])); - // log("putexp", parts[1], parts[2]); + long ttlSeconds = Long.parseLong(parts[3]); + long expiryTime = System.currentTimeMillis() + (ttlSeconds); + map.putexp(parts[1], parts[2], ttlSeconds); + ttlManager.schedule(parts[1], expiryTime); + log("putexp", parts[1], parts[2] + " " + ttlSeconds); System.out.println("OK"); } else { - System.out.println("Usage: put "); + System.out.println("Usage: putexp "); } break; case "get": @@ -168,12 +177,14 @@ public static void interactiveMode() { break; case "help": System.out.println( - "Commands: putexp ,put , get , del , show, count, exit" - ); + "Commands: putexp ,put , get , del , show, count, exit"); break; case "exit": case "quit": - if (logWriter != null) logWriter.close(); + if (logWriter != null) + logWriter.close(); + if (ttlManager != null) + ttlManager.shutdown(); System.out.println("Goodbye!"); return; default: diff --git a/src/main/java/TTLManager.java b/src/main/java/TTLManager.java new file mode 100644 index 0000000..2440589 --- /dev/null +++ b/src/main/java/TTLManager.java @@ -0,0 +1,59 @@ +import java.util.concurrent.DelayQueue; + +public class TTLManager { + // delayqueue is also a priorityqueue only but with default expiry for time and + // the take() method blocking until an entry is past it's ttl that prevents me + // from checking explicitly when to run the background thread to check the top + // of the queue.; + // delayqueue uses expiryEntry to check the delayed; + private final DelayQueue expiryQueue = new DelayQueue<>(); + private final Jmap map; + private volatile boolean running = true; + private Thread cleanupThread; + + public TTLManager(Jmap map) { + this.map = map; + startCleanupThread(); + } + + public void schedule(String key, long expiryTimeMillis) { + expiryQueue.put(new ExpiryEntry(key, expiryTimeMillis)); + } + + private void startCleanupThread() { + cleanupThread = new Thread(() -> { + while (running) { + try { + // Blocks until an entry is ready to expire + ExpiryEntry entry = expiryQueue.take(); + + // key might have been updated/deleted) + Long currentExpiry = map.getExpiry(entry.getKey()); + if (currentExpiry != null && currentExpiry == entry.getExpiryTime()) { + try { + map.remove(entry.getKey()); + System.out.println("[TTL] Expired and removed key: " + entry.getKey()); + } catch (IllegalArgumentException e) { + // Key was already removed, ignore + } + } + // If expiry doesn't match, the key was updated — ignore stale entry + + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + }, "TTL-Cleanup-Thread"); + + cleanupThread.setDaemon(true); + cleanupThread.start(); + } + + public void shutdown() { + running = false; + if (cleanupThread != null) { + cleanupThread.interrupt(); + } + } +} diff --git a/src/test/java/TTLTest.java b/src/test/java/TTLTest.java new file mode 100644 index 0000000..81f308d --- /dev/null +++ b/src/test/java/TTLTest.java @@ -0,0 +1,171 @@ +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +public class TTLTest { + + public static void main(String[] args) throws InterruptedException { + testBasicTTL(); + testMultipleTTLs(); + testTTLUpdateScenario(); + testConcurrentTTLOperations(); + System.out.println("All TTL tests passed!"); + } + + /** + * Test basic TTL expiration + */ + public static void testBasicTTL() throws InterruptedException { + System.out.println("\n=== Testing Basic TTL ==="); + Jmap map = new Jmap<>(0.75f); + TTLManager ttlManager = new TTLManager(map); + + // Put key with 2 second TTL + long expiryTime = System.currentTimeMillis() + 2000; + map.putexp("testkey", "testvalue", 2000); + ttlManager.schedule("testkey", expiryTime); + + // Should exist immediately + assert map.get("testkey").equals("testvalue") : "Key should exist immediately"; + System.out.println("✓ Key exists immediately after putexp"); + + // Wait 1 second - should still exist + Thread.sleep(1000); + assert map.get("testkey").equals("testvalue") : "Key should exist after 1s"; + System.out.println("✓ Key exists after 1s"); + + // Wait 2 more seconds - should be expired + Thread.sleep(2500); + assert map.get("testkey") == null : "Key should be expired after 3.5s"; + System.out.println("✓ Key expired after 3.5s"); + + ttlManager.shutdown(); + } + + /** + * Test multiple keys with different TTLs + */ + public static void testMultipleTTLs() throws InterruptedException { + System.out.println("\n=== Testing Multiple TTLs ==="); + Jmap map = new Jmap<>(0.75f); + TTLManager ttlManager = new TTLManager(map); + + long now = System.currentTimeMillis(); + + // Key1: 1 second TTL + map.putexp("key1", "value1", 1000); + ttlManager.schedule("key1", now + 1000); + + // Key2: 3 second TTL + map.putexp("key2", "value2", 3000); + ttlManager.schedule("key2", now + 3000); + + // Key3: 5 second TTL + map.putexp("key3", "value3", 5000); + ttlManager.schedule("key3", now + 5000); + + // All should exist initially + assert map.get("key1") != null && map.get("key2") != null && map.get("key3") != null; + System.out.println("✓ All keys exist initially"); + + // After 1.5s: key1 expired, key2 and key3 exist + Thread.sleep(1500); + assert map.get("key1") == null : "key1 should be expired"; + assert map.get("key2") != null : "key2 should exist"; + assert map.get("key3") != null : "key3 should exist"; + System.out.println("✓ key1 expired, key2 and key3 exist after 1.5s"); + + // After 3.5s total: key1 and key2 expired, key3 exists + Thread.sleep(2000); + assert map.get("key1") == null : "key1 should be expired"; + assert map.get("key2") == null : "key2 should be expired"; + assert map.get("key3") != null : "key3 should exist"; + System.out.println("✓ key1 and key2 expired, key3 exists after 3.5s"); + + // After 5.5s total: all expired + Thread.sleep(2000); + assert map.get("key1") == null && map.get("key2") == null && map.get("key3") == null; + System.out.println("✓ All keys expired after 5.5s"); + + ttlManager.shutdown(); + } + + /** + * Test TTL update scenario - old entry in queue should be ignored + */ + public static void testTTLUpdateScenario() throws InterruptedException { + System.out.println("\n=== Testing TTL Update Scenario ==="); + Jmap map = new Jmap<>(0.75f); + TTLManager ttlManager = new TTLManager(map); + + long now = System.currentTimeMillis(); + + // Initial: 2 second TTL + map.putexp("updatekey", "value1", 2000); + ttlManager.schedule("updatekey", now + 2000); + + // Wait 1 second, then update with 4 second TTL from now + Thread.sleep(1000); + long newExpiry = System.currentTimeMillis() + 4000; + map.putexp("updatekey", "value2", 4000); + ttlManager.schedule("updatekey", newExpiry); + + // After original 2s would have expired, key should still exist + Thread.sleep(1500); // 2.5s total + assert map.get("updatekey") != null : "Key should still exist after update"; + assert map.get("updatekey").equals("value2") : "Value should be updated"; + System.out.println("✓ Key exists after TTL update (stale entry ignored)"); + + // Should expire after the new TTL + Thread.sleep(3000); // 5.5s total + assert map.get("updatekey") == null : "Key should be expired after new TTL"; + System.out.println("✓ Key expired after new TTL period"); + + ttlManager.shutdown(); + } + + /** + * Test concurrent TTL operations + */ + public static void testConcurrentTTLOperations() throws InterruptedException { + System.out.println("\n=== Testing Concurrent TTL Operations ==="); + Jmap map = new Jmap<>(0.75f); + TTLManager ttlManager = new TTLManager(map); + + int numKeys = 100; + CountDownLatch latch = new CountDownLatch(numKeys); + + // Add multiple keys with TTL concurrently + for (int i = 0; i < numKeys; i++) { + final int keyNum = i; + new Thread(() -> { + try { + long ttl = 1000 + (keyNum % 3) * 1000; // 1-3 second TTL + long expiry = System.currentTimeMillis() + ttl; + map.putexp("concurrentkey" + keyNum, "value" + keyNum, ttl); + ttlManager.schedule("concurrentkey" + keyNum, expiry); + } finally { + latch.countDown(); + } + }).start(); + } + + latch.await(); // Wait for all puts to complete + System.out.println("✓ All concurrent puts completed"); + + // Wait for all to expire + Thread.sleep(4500); + + // Check all are expired + int existingCount = 0; + for (int i = 0; i < numKeys; i++) { + if (map.get("concurrentkey" + i) != null) { + existingCount++; + } + } + + assert existingCount == 0 : "All keys should be expired, but " + existingCount + " still exist"; + System.out.println("✓ All concurrent keys expired properly"); + + ttlManager.shutdown(); + } +} \ No newline at end of file From 31be7ff71a747cb95cbedbc0a58fc7ee51529a90 Mon Sep 17 00:00:00 2001 From: Rishabh Date: Tue, 26 May 2026 21:52:50 +0530 Subject: [PATCH 2/3] fixed bugs --- README.md | 11 ++++++- src/main/java/ExpiryEntry.java | 8 +++--- src/main/java/Jmap.java | 52 ++++++++++++++-------------------- src/main/java/TTLManager.java | 14 ++++----- 4 files changed, 42 insertions(+), 43 deletions(-) diff --git a/README.md b/README.md index d5e8426..9661179 100644 --- a/README.md +++ b/README.md @@ -16,4 +16,13 @@ This project is a custom implementation of a thread-safe HashMap in Java. It is - `DELETE`: To remove a key-value pair. -mvn exec:java -Dexec.mainClass="KVStore" \ No newline at end of file +mvn exec:java -Dexec.mainClass="KVStore" + +## Known Issues + +- **Thread Safety Violation in `get()`**: Modifies the linked list with only a read lock when lazily removing expired keys. +- **Count Inconsistency**: When `get()` dynamically removes expired keys, it fails to decrement `node_ct`. +- **Stale Lock Reference After Resize**: The `put` & `putexp` methods reacquire an old bucket lock (via a stale local variable) after a resize finishes, instead of looking up the new lock array. +- **Inconsistent TTL Calculation**: `putexp` applies absolute timestamps to updated entries, but relative TTLs to newly inserted entries (unless handled inside the `Jnode` constructor, which breaks encapsulation). +- **`resize_put` Encapsulation**: The method is marked `public` but skips bucket locking; it should be `private`. +- **No TTL Check on `remove()`**: Forced deletions behave identically for alive keys and logically expired ones. diff --git a/src/main/java/ExpiryEntry.java b/src/main/java/ExpiryEntry.java index b3fafcf..f044675 100644 --- a/src/main/java/ExpiryEntry.java +++ b/src/main/java/ExpiryEntry.java @@ -1,16 +1,16 @@ import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; -public class ExpiryEntry implements Delayed { - private final String key; +public class ExpiryEntry implements Delayed { + private final K key; private final long expiryTime; // absolute time in millis - public ExpiryEntry(String key, long expiryTime) { + public ExpiryEntry(K key, long expiryTime) { this.key = key; this.expiryTime = expiryTime; } - public String getKey() { + public K getKey() { return key; } diff --git a/src/main/java/Jmap.java b/src/main/java/Jmap.java index 2d2e427..c694e03 100644 --- a/src/main/java/Jmap.java +++ b/src/main/java/Jmap.java @@ -5,10 +5,10 @@ public class Jmap { - private Jnode[] table; + private volatile Jnode[] table; private final AtomicInteger size = new AtomicInteger(16); private float resize_threshold = 0.75f; // so when 75% of the capacity is filled i resize the thing right; - public static volatile AtomicInteger node_ct = new AtomicInteger(0); + public AtomicInteger node_ct = new AtomicInteger(0); private ReadWriteLock[] rwLocks; private ReadWriteLock globalRwLock = new ReentrantReadWriteLock(); @@ -48,10 +48,6 @@ private int hashFunction(K key, int mapCapacity) { return h & (mapCapacity - 1); } - // how do i efficiently put a value with ttl? - // do i overload the method but that would result in duplicate code; - // do i put a condition - public void put(K key, V value) { readLock.lock(); int tableIndex = hashFunction(key, size.get()); @@ -178,14 +174,14 @@ public void remove(K key) { Jnode current = table[tableIndex]; Jnode prev = null; if (current == null) { - throw new IllegalArgumentException("Key not found: " + key); + return; } while (current != null && !current.key.equals(key)) { prev = current; current = current.next; } if (current == null) { - throw new IllegalArgumentException("Key not found: " + key); + return; // key not found, nothing to remove; } if (prev == null) { // if it's the first node only;; @@ -218,14 +214,8 @@ public V get(K key) { if (value.key.equals(key) && (value.ttl == -1 || System.currentTimeMillis() <= value.ttl)) { return value.value; } else if (value.key.equals(key) && value.ttl != -1 && System.currentTimeMillis() > value.ttl) { - // will have to manually remove the entry, can't use remove() bcoz it'll - // deadlock; - if (prev == null) { - table[tableIndex] = value.next; - } else { - prev.next = value.next; - } - System.out.println("removed old entry:" + value.key + " " + value.value); + + //dont' have to do anything if the entry is expired the ttlmanager will take care of the cleanup and the get will just return null for expired entries, so i just return null here and let the ttl manager do it; return null; } else { prev = value; @@ -265,23 +255,16 @@ public Long getExpiry(K key) { } } - public void resize_put(K key, V value, Jnode[] newTable) { - int tableIndex = hashFunction(key, newTable.length); - Jnode node = new Jnode<>(tableIndex, key, value); + public void resize_put(Jnode node, Jnode[] newTable) { + int tableIndex = hashFunction(node.key, newTable.length); + if (newTable[tableIndex] == null) { newTable[tableIndex] = node; - node_ct.getAndIncrement(); + } else { Jnode currHeadNode = newTable[tableIndex]; - while (currHeadNode != null) { - if (currHeadNode.next != null) { - currHeadNode = currHeadNode.next; - } else { - currHeadNode.next = node; - node_ct.getAndIncrement(); - return; - } - } + node.next = currHeadNode; + newTable[tableIndex] = node; } } @@ -289,14 +272,21 @@ public void resize_put(K key, V value, Jnode[] newTable) { private void resizeJmap() { // so a completely new array with twice the size is required now int newSize = size.get() * 2; - node_ct.set(0); Jnode[] newTable = (Jnode[]) new Jnode[newSize]; for (int j = 0; j < size.get(); j++) { Jnode currNode = table[j]; while (currNode != null) { - resize_put(currNode.key, currNode.value, newTable); + if (currNode.ttl != -2) { + resize_put(currNode, newTable);// after this the currnode is added to the new table and the link to + // the old next is removed, so i need make the next node the current + // head of the linked list in the old table and then move on to the + // next node; + } + Jnode temp = currNode; currNode = currNode.next; + temp.next = null; + } } diff --git a/src/main/java/TTLManager.java b/src/main/java/TTLManager.java index 2440589..dd1f13f 100644 --- a/src/main/java/TTLManager.java +++ b/src/main/java/TTLManager.java @@ -1,23 +1,23 @@ import java.util.concurrent.DelayQueue; -public class TTLManager { +public class TTLManager { // delayqueue is also a priorityqueue only but with default expiry for time and // the take() method blocking until an entry is past it's ttl that prevents me // from checking explicitly when to run the background thread to check the top // of the queue.; // delayqueue uses expiryEntry to check the delayed; - private final DelayQueue expiryQueue = new DelayQueue<>(); - private final Jmap map; + private final DelayQueue> expiryQueue = new DelayQueue<>(); + private final Jmap map; private volatile boolean running = true; private Thread cleanupThread; - public TTLManager(Jmap map) { + public TTLManager(Jmap map) { this.map = map; startCleanupThread(); } - public void schedule(String key, long expiryTimeMillis) { - expiryQueue.put(new ExpiryEntry(key, expiryTimeMillis)); + public void schedule(K key, long expiryTimeMillis) { + expiryQueue.put(new ExpiryEntry(key, expiryTimeMillis)); } private void startCleanupThread() { @@ -25,7 +25,7 @@ private void startCleanupThread() { while (running) { try { // Blocks until an entry is ready to expire - ExpiryEntry entry = expiryQueue.take(); + ExpiryEntry entry = expiryQueue.take(); // key might have been updated/deleted) Long currentExpiry = map.getExpiry(entry.getKey()); From 164f2b6bbf1c5fdd32ebe4ca1c4e5d8b5970988f Mon Sep 17 00:00:00 2001 From: Rishabh Date: Fri, 29 May 2026 19:54:45 +0530 Subject: [PATCH 3/3] shifted to stamped locks and benchmarked --- README.md | 66 +++-- dependency-reduced-pom.xml | 85 +++++++ pom.xml | 56 ++++- src/main/java/{ => cache}/ExpiryEntry.java | 1 + src/main/java/{ => cache}/Jmap.java | 265 ++++++++++----------- src/main/java/cache/JmapBenchmark.java | 92 +++++++ src/main/java/{ => cache}/Jnode.java | 1 + src/main/java/{ => cache}/KVStore.java | 22 +- src/main/java/cache/StressTest.java | 87 +++++++ src/main/java/{ => cache}/TTLManager.java | 3 +- src/main/logs/logs.txt | 1 + src/test/java/TTLTest.java | 3 + 12 files changed, 505 insertions(+), 177 deletions(-) create mode 100644 dependency-reduced-pom.xml rename src/main/java/{ => cache}/ExpiryEntry.java (98%) rename src/main/java/{ => cache}/Jmap.java (50%) create mode 100644 src/main/java/cache/JmapBenchmark.java rename src/main/java/{ => cache}/Jnode.java (98%) rename src/main/java/{ => cache}/KVStore.java (95%) create mode 100644 src/main/java/cache/StressTest.java rename src/main/java/{ => cache}/TTLManager.java (94%) diff --git a/README.md b/README.md index 9661179..9934d5c 100644 --- a/README.md +++ b/README.md @@ -1,28 +1,58 @@ -# Custom HashMap +# Concurrent Key-Value Store (Jmap) -This project is a custom implementation of a thread-safe HashMap in Java. It is designed to be a learning exercise to understand the internal workings of a HashMap and multithreading in java. +A custom, high-performance Concurrent HashMap implementation in Java with Time-To-Live (TTL) support, extensive concurrency strategies, and a built-in interactive Command Line Interface. ## Features -- **Thread-Safe**: The `Jmap` class is thread-safe. It uses bucket-level locking to allow multiple threads to access the map concurrently. This is more efficient than locking the entire map for every operation. -- **Automatic Resizing**: The map automatically resizes itself when the number of elements exceeds the load factor. This is done to maintain a constant time complexity for the basic operations like `put`, `get`, and `remove`. -- **Key-Value Store**: The `KVStore` class provides a simple key-value store that uses the `Jmap` as its underlying data structure. It provides an interactive command-line interface to perform `put`, `get`, and `delete` operations. -- **Persistence**: The `KVStore` class also provides a simple persistence mechanism. It logs all the `put` and `delete` operations to a log file. When the application is started, it rebuilds the in-memory key-value store from the log file. +- **Custom Concurrent Map (`Jmap`)**: Implements a highly concurrent hash map using bucket-level `ReentrantReadWriteLock`s to minimize thread contention down to the specific bucket level. +- **Optimistic Reads**: Utilizes Java's `StampedLock` for fast optimistic reads, scaling up throughput. When validation fails (like during a concurrent resize), it gracefully falls back to secure read locks. +- **Dynamic Resizing**: Automatically handles resizing and re-hashing with minimal disruption when the load factor threshold is reached (default 0.75). +- **Time-To-Live (TTL) Expiration**: Attach an expiration time to any key using the `putexp` command. A background `TTLManager` daemon thread efficiently cleans up expired keys using a `DelayQueue`. -## Methods +## Project Structure -- `PUT`: To add a new key-value pair. -- `GET`: To retrieve the value for a given key. -- `DELETE`: To remove a key-value pair. +- `src/main/java/cache/Jmap.java`: The core concurrent hash map implementation, featuring fine-grained locking and optimistic reads. +- `src/main/java/cache/Jnode.java`: Represents a node/bucket entry in the map. +- `src/main/java/cache/KVStore.java`: The main application class providing the interactive CLI. +- `src/main/java/cache/TTLManager.java`: Manages key expiration and cleanup using a background daemon thread. +- `src/main/java/cache/ExpiryEntry.java`: A supporting class used by `DelayQueue` to track the expiration time of keys. +- `src/main/java/cache/JmapBenchmark.java`: JMH benchmarking setup to compare throughput running with 1, 8, and 16 concurrent threads. +- `src/main/java/cache/StressTest.java`: Multi-threaded integration tests running intensive concurrent operations. +- `src/test/java/cache/TTLTest.java`: Unit tests ensuring correct TTL behavior. +## Usage -mvn exec:java -Dexec.mainClass="KVStore" +You can run the interactive CLI by executing the `KVStore` main class. -## Known Issues +### CLI Commands -- **Thread Safety Violation in `get()`**: Modifies the linked list with only a read lock when lazily removing expired keys. -- **Count Inconsistency**: When `get()` dynamically removes expired keys, it fails to decrement `node_ct`. -- **Stale Lock Reference After Resize**: The `put` & `putexp` methods reacquire an old bucket lock (via a stale local variable) after a resize finishes, instead of looking up the new lock array. -- **Inconsistent TTL Calculation**: `putexp` applies absolute timestamps to updated entries, but relative TTLs to newly inserted entries (unless handled inside the `Jnode` constructor, which breaks encapsulation). -- **`resize_put` Encapsulation**: The method is marked `public` but skips bucket locking; it should be `private`. -- **No TTL Check on `remove()`**: Forced deletions behave identically for alive keys and logically expired ones. +- `put `: Insert or update a key with a value. +- `putexp `: Insert or update a key with a value and a Time-To-Live (TTL) in seconds. +- `get `: Retrieve the value associated with a key. +- `del ` (or `remove `): Delete a key from the store. +- `show` (or `display`): Print the internal state and structure of the map. +- `count` (or `size`): Print the total number of non-expired nodes in the map. +- `help`: Display the list of available commands. +- `exit` (or `quit`): Safely shut down the CLI and background threads. + +## Build and Run + +This project uses Maven for dependency management and builds. Built on Java 21, packing standard tools like JUnit 5 and JMH. + +```bash +mvn clean compile +mvn exec:java -Dexec.mainClass="cache.KVStore" +``` + +### Running Benchmarks + +Because the program implements JMH benchmarks, you can package and run them locally to observe throughput comparison against `java.util.concurrent.ConcurrentHashMap`: + +```bash +mvn clean install +java -jar target/benchmarks.jar +``` +Global ReentrantReadWriteLock became a bottleneck as thread count increased. Although multiple readers can hold the read lock concurrently, acquiring and releasing the read lock is not free. Internally, the lock maintains shared state (such as reader counts) that is updated atomically. This lock state resides in a cache line that may be cached by multiple CPU cores. When a thread acquires or releases the read lock, its core must obtain exclusive ownership of that cache line before modifying it. This causes cache coherence traffic as ownership of the cache line moves between cores. Under high read concurrency, many threads contend on the same lock metadata, causing frequent cache-line bouncing and reducing scalability. As thread count increases, the overhead of maintaining the shared reader count becomes significant even though the protected data itself is only being read. +Future scope: + +Might move fom one single global lock to sharding the entire map into smaller pieces and having locks for those smaller shards because stampedlock doesn't always try to acquire a lock but it can also still fail when no. of writes are more. \ No newline at end of file diff --git a/dependency-reduced-pom.xml b/dependency-reduced-pom.xml new file mode 100644 index 0000000..b1e5474 --- /dev/null +++ b/dependency-reduced-pom.xml @@ -0,0 +1,85 @@ + + + 4.0.0 + customhashmap + customhashmap + 1.0-SNAPSHOT + + + + maven-shade-plugin + 3.5.0 + + + package + + shade + + + + + benchmarks + + + org.openjdk.jmh.Main + + + + + + + maven-compiler-plugin + 3.13.0 + + 21 + + + org.openjdk.jmh + jmh-generator-annprocess + 1.37 + + + + + + org.codehaus.mojo + exec-maven-plugin + 3.1.0 + + cache.StressTest + + + + maven-surefire-plugin + 3.2.5 + + + + + + org.junit.jupiter + junit-jupiter + 5.10.2 + test + + + junit-jupiter-api + org.junit.jupiter + + + junit-jupiter-params + org.junit.jupiter + + + junit-jupiter-engine + org.junit.jupiter + + + + + + UTF-8 + 21 + 21 + + diff --git a/pom.xml b/pom.xml index e1b03b8..e05d644 100644 --- a/pom.xml +++ b/pom.xml @@ -2,10 +2,9 @@ xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 - http://maven.apache.org/xsd/maven-4.0.0.xsd" -> - 4.0.0 + http://maven.apache.org/xsd/maven-4.0.0.xsd"> + 4.0.0 customhashmap customhashmap 1.0-SNAPSHOT @@ -23,27 +22,63 @@ 5.10.2 test + + org.openjdk.jmh + jmh-core + 1.37 + + + org.openjdk.jmh + jmh-generator-annprocess + 1.37 + - + + org.apache.maven.plugins + maven-shade-plugin + 3.5.0 + + + package + + shade + + + + + benchmarks + + + org.openjdk.jmh.Main + + + + + org.apache.maven.plugins maven-compiler-plugin - 3.8.1 + 3.13.0 - 17 + 21 + + + org.openjdk.jmh + jmh-generator-annprocess + 1.37 + + - - org.codehaus.mojo exec-maven-plugin 3.1.0 - KVStore + cache.StressTest @@ -53,4 +88,5 @@ - + + \ No newline at end of file diff --git a/src/main/java/ExpiryEntry.java b/src/main/java/cache/ExpiryEntry.java similarity index 98% rename from src/main/java/ExpiryEntry.java rename to src/main/java/cache/ExpiryEntry.java index f044675..7e9ed7a 100644 --- a/src/main/java/ExpiryEntry.java +++ b/src/main/java/cache/ExpiryEntry.java @@ -1,3 +1,4 @@ +package cache; import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; diff --git a/src/main/java/Jmap.java b/src/main/java/cache/Jmap.java similarity index 50% rename from src/main/java/Jmap.java rename to src/main/java/cache/Jmap.java index c694e03..0a5bec0 100644 --- a/src/main/java/Jmap.java +++ b/src/main/java/cache/Jmap.java @@ -1,19 +1,19 @@ +package cache; + import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.locks.StampedLock; public class Jmap { private volatile Jnode[] table; private final AtomicInteger size = new AtomicInteger(16); - private float resize_threshold = 0.75f; // so when 75% of the capacity is filled i resize the thing right; + private float resize_threshold = 0.75f; public AtomicInteger node_ct = new AtomicInteger(0); - private ReadWriteLock[] rwLocks; - private ReadWriteLock globalRwLock = new ReentrantReadWriteLock(); - private Lock resizeLock = globalRwLock.writeLock(); // only acquire this at the time of resizing; - private Lock readLock = globalRwLock.readLock(); // only acquire this at the time of resizing; + private volatile ReadWriteLock[] rwLocks; + private final StampedLock stampedLock = new StampedLock(); @SuppressWarnings("unchecked") public Jmap() { @@ -34,12 +34,8 @@ public Jmap(float threshold) { } } - // now every time a new node is placed i check if resizing is required or not; - private int hashFunction(K key, int mapCapacity) { - // so when a key is provided it returns a hash value; int h = Math.abs(key.hashCode()); - // same as concurrentHahsMap; h += (h << 15) ^ 0xffffcd7d; h ^= (h >>> 10); h += (h << 3); @@ -49,16 +45,16 @@ private int hashFunction(K key, int mapCapacity) { } public void put(K key, V value) { - readLock.lock(); + // + long stamp = stampedLock.readLock(); int tableIndex = hashFunction(key, size.get()); ReadWriteLock bucketLock = rwLocks[tableIndex]; bucketLock.writeLock().lock(); + boolean needResize = false; try { boolean newNodePlaced = false; Jnode node = new Jnode<>(tableIndex, key, value); - // System.out.println(Thread.currentThread().getName() + " got lock " + - // tableIndex); if (table[tableIndex] == null) { table[tableIndex] = node; node_ct.getAndIncrement(); @@ -82,42 +78,34 @@ public void put(K key, V value) { } } } - if (newNodePlaced) { - boolean resizeRequired = (node_ct.get() >= size.get() * resize_threshold); - if (resizeRequired) { - readLock.unlock(); - bucketLock.writeLock().unlock(); - resizeLock.lock(); // checking again if resize is required - try { - if ((node_ct.get() >= size.get() * resize_threshold)) { - // System.out.println("resize triggereed"); - resizeJmap(); - } - } finally { - resizeLock.unlock(); - readLock.lock(); // reacquired this because of the release outside; - bucketLock.writeLock().lock(); - } - } - } + needResize = newNodePlaced && (node_ct.get() >= size.get() * resize_threshold); } finally { bucketLock.writeLock().unlock(); - readLock.unlock(); + stampedLock.unlockRead(stamp); + } + + if (needResize) { + long writeStamp = stampedLock.writeLock(); + try { + if (node_ct.get() >= size.get() * resize_threshold) { + resizeJmap(); + } + } finally { + stampedLock.unlockWrite(writeStamp); + } } } - // put/putexp on an existing key overwrites the ttl of the prev key; public void putexp(K key, V value, long ttl) { - readLock.lock(); + long stamp = stampedLock.readLock(); int tableIndex = hashFunction(key, size.get()); ReadWriteLock bucketLock = rwLocks[tableIndex]; bucketLock.writeLock().lock(); + boolean needResize = false; try { boolean newNodePlaced = false; Jnode node = new Jnode<>(tableIndex, key, value, ttl); - // System.out.println(Thread.currentThread().getName() + " got lock " + - // tableIndex); if (table[tableIndex] == null) { table[tableIndex] = node; node_ct.getAndIncrement(); @@ -141,155 +129,159 @@ public void putexp(K key, V value, long ttl) { } } } - if (newNodePlaced) { - boolean resizeRequired = (node_ct.get() >= size.get() * resize_threshold); - if (resizeRequired) { - readLock.unlock(); - bucketLock.writeLock().unlock(); - resizeLock.lock(); // checking again if resize is required - try { - if ((node_ct.get() >= size.get() * resize_threshold)) { - // System.out.println("resize triggereed"); - resizeJmap(); - } - } finally { - resizeLock.unlock(); - readLock.lock(); // reacquired this because of the release outside; - bucketLock.writeLock().lock(); - } - } - } + needResize = newNodePlaced && (node_ct.get() >= size.get() * resize_threshold); } finally { bucketLock.writeLock().unlock(); - readLock.unlock(); + stampedLock.unlockRead(stamp); + } + + if (needResize) { + long writeStamp = stampedLock.writeLock(); + try { + if (node_ct.get() >= size.get() * resize_threshold) { + resizeJmap(); + } + } finally { + stampedLock.unlockWrite(writeStamp); + } } } public void remove(K key) { - readLock.lock(); + long stamp = stampedLock.readLock(); int tableIndex = hashFunction(key, size.get()); ReadWriteLock bucketLock = rwLocks[tableIndex]; bucketLock.writeLock().lock(); try { Jnode current = table[tableIndex]; Jnode prev = null; - if (current == null) { - return; - } + if (current == null) return; + while (current != null && !current.key.equals(key)) { prev = current; current = current.next; } - if (current == null) { - return; // key not found, nothing to remove; - } + if (current == null) return; + if (prev == null) { - // if it's the first node only;; table[tableIndex] = current.next; } else { - prev.next = current.next; // deleted the curr one gc removes it; + prev.next = current.next; } - node_ct.decrementAndGet(); } finally { bucketLock.writeLock().unlock(); - readLock.unlock(); + stampedLock.unlockRead(stamp); } } public V get(K key) { - // so this is a read i just need to acquire the global read lock and the bucket - // level read lock; - // blocks incase a resize is happening; - readLock.lock(); - int tableIndex = hashFunction(key, size.get()); - ReadWriteLock bucketLock = rwLocks[tableIndex]; - bucketLock.readLock().lock(); - // prevents reads and writes on same bucket at the saem time , and it allows - // multiple reads to happen together, previously reads would've blocked. - try { - Jnode value = table[tableIndex]; - Jnode prev = null; - while (value != null) { - if (value.key.equals(key) && (value.ttl == -1 || System.currentTimeMillis() <= value.ttl)) { - return value.value; - } else if (value.key.equals(key) && value.ttl != -1 && System.currentTimeMillis() > value.ttl) { - - //dont' have to do anything if the entry is expired the ttlmanager will take care of the cleanup and the get will just return null for expired entries, so i just return null here and let the ttl manager do it; - return null; - } else { - prev = value; - value = value.next; + while (true) { + long stamp = stampedLock.tryOptimisticRead(); + int tableIndex = hashFunction(key, size.get()); + Jnode[] localTable = table; + ReadWriteLock[] localLocks = rwLocks; + boolean optimisticReadFailed = false; + //if the the validation fails it means that resizing happened because the stampled write lock can only cause our stamp to be invalidated; + + + if (!stampedLock.validate(stamp)) { + stamp = stampedLock.readLock();//now we try with a readlock to make sure that a resize doesn't happen while we are trying to acquire the bucket lock; + optimisticReadFailed = true; + tableIndex = hashFunction(key, size.get()); + localTable = table; + localLocks = rwLocks; + }//the resize has caused the buckets to be changed; + + ReadWriteLock bucketLock = localLocks[tableIndex]; + bucketLock.readLock().lock(); + + + // resize happened between our optimistic validate and bucket lock acquisition + if (!optimisticReadFailed && !stampedLock.validate(stamp)) { + bucketLock.readLock().unlock(); + continue; + } + + try { + Jnode node = localTable[tableIndex]; + while (node != null) { + if (node.key.equals(key) && (node.ttl == -1 || System.currentTimeMillis() <= node.ttl)) { + return node.value; + } else if (node.key.equals(key) && node.ttl != -1 && System.currentTimeMillis() > node.ttl) { + return null; + } else { + node = node.next; + } } + return null; + } finally { + bucketLock.readLock().unlock(); + if (optimisticReadFailed) stampedLock.unlockRead(stamp); } - return null; - } finally { - /// always executes; - bucketLock.readLock().unlock(); - readLock.unlock(); } } - /** - * Returns the expiry time (absolute millis) for a key, or null if key doesn't - * exist or has no TTL. - * Used by TTLManager to validate stale entries before deletion. - */ public Long getExpiry(K key) { - readLock.lock(); - int tableIndex = hashFunction(key, size.get()); - ReadWriteLock bucketLock = rwLocks[tableIndex]; - bucketLock.readLock().lock(); - try { - Jnode node = table[tableIndex]; - while (node != null) { - if (node.key.equals(key)) { - return node.ttl == -1 ? null : node.ttl; + while (true) { + long stamp = stampedLock.tryOptimisticRead(); + int tableIndex = hashFunction(key, size.get()); + Jnode[] localTable = table; + ReadWriteLock[] localLocks = rwLocks; + boolean optimisticReadFailed = false; + + if (!stampedLock.validate(stamp)) { + stamp = stampedLock.readLock(); + optimisticReadFailed = true; + tableIndex = hashFunction(key, size.get()); + localTable = table; + localLocks = rwLocks; + } + + ReadWriteLock bucketLock = localLocks[tableIndex]; + bucketLock.readLock().lock(); + + if (!optimisticReadFailed && !stampedLock.validate(stamp)) { + bucketLock.readLock().unlock(); + continue; + } + + try { + Jnode node = localTable[tableIndex]; + while (node != null) { + if (node.key.equals(key)) { + return node.ttl == -1 ? null : node.ttl; + } + node = node.next; } - node = node.next; + return null; + } finally { + bucketLock.readLock().unlock(); + if (optimisticReadFailed) stampedLock.unlockRead(stamp); } - return null; - } finally { - bucketLock.readLock().unlock(); - readLock.unlock(); } } public void resize_put(Jnode node, Jnode[] newTable) { int tableIndex = hashFunction(node.key, newTable.length); - - if (newTable[tableIndex] == null) { - newTable[tableIndex] = node; - - } else { - Jnode currHeadNode = newTable[tableIndex]; - node.next = currHeadNode; - newTable[tableIndex] = node; - } + node.next = newTable[tableIndex]; + newTable[tableIndex] = node; } @SuppressWarnings("unchecked") private void resizeJmap() { - // so a completely new array with twice the size is required now int newSize = size.get() * 2; Jnode[] newTable = (Jnode[]) new Jnode[newSize]; for (int j = 0; j < size.get(); j++) { Jnode currNode = table[j]; - while (currNode != null) { + Jnode nextNode = currNode.next; if (currNode.ttl != -2) { - resize_put(currNode, newTable);// after this the currnode is added to the new table and the link to - // the old next is removed, so i need make the next node the current - // head of the linked list in the old table and then move on to the - // next node; + resize_put(currNode, newTable); } - Jnode temp = currNode; - currNode = currNode.next; - temp.next = null; - + currNode = nextNode; } } - this.table = newTable; this.size.set(newSize); this.rwLocks = new ReentrantReadWriteLock[size.get()]; @@ -301,22 +293,19 @@ private void resizeJmap() { @Override public String toString() { StringBuilder str = new StringBuilder(); - for (int j = 0; j < size.get(); j++) { if (table[j] != null) { Jnode tempNode = table[j]; - String formattedString = String.format("[%d] ->", j); - str.append(formattedString); + str.append(String.format("[%d] ->", j)); while (tempNode != null) { - // append every node to str; str.append(tempNode).append("->"); tempNode = tempNode.next; } str.append("\n"); } else { - str.append("[").append(j).append("]->").append("null").append("\n"); + str.append("[").append(j).append("]->null\n"); } } return str.toString(); } -} +} \ No newline at end of file diff --git a/src/main/java/cache/JmapBenchmark.java b/src/main/java/cache/JmapBenchmark.java new file mode 100644 index 0000000..f6646b2 --- /dev/null +++ b/src/main/java/cache/JmapBenchmark.java @@ -0,0 +1,92 @@ +package cache; + + + +import org.openjdk.jmh.annotations.*; +import org.openjdk.jmh.infra.Blackhole; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; + +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@State(Scope.Benchmark) +@Warmup(iterations = 3, time = 1) +@Measurement(iterations = 5, time = 1) +@Fork(1) +public class JmapBenchmark { + + private Jmap jmap; + private ConcurrentHashMap chm; + + private static final int KEY_SPACE = 10_000; + + @Setup + public void setup() { + jmap = new Jmap<>(); + chm = new ConcurrentHashMap<>(); + for (int i = 0; i < KEY_SPACE; i++) { + jmap.put("key" + i, "val" + i); + chm.put("key" + i, "val" + i); + } + } + + @Benchmark + @Threads(1) + public void jmapSingleThread(Blackhole bh) { + int key = ThreadLocalRandom.current().nextInt(KEY_SPACE); + bh.consume(jmap.get("key" + key)); + } + + @Benchmark + @Threads(1) + public void chmSingleThread(Blackhole bh) { + int key = ThreadLocalRandom.current().nextInt(KEY_SPACE); + bh.consume(chm.get("key" + key)); + } + + @Benchmark + @Threads(8) + public void jmap8Threads(Blackhole bh) { + int key = ThreadLocalRandom.current().nextInt(KEY_SPACE); + if (ThreadLocalRandom.current().nextInt(10) < 9) { + bh.consume(jmap.get("key" + key)); + } else { + jmap.put("key" + key, "val" + key); + } + } + + @Benchmark + @Threads(8) + public void chm8Threads(Blackhole bh) { + int key = ThreadLocalRandom.current().nextInt(KEY_SPACE); + if (ThreadLocalRandom.current().nextInt(10) < 9) { + bh.consume(chm.get("key" + key)); + } else { + chm.put("key" + key, "val" + key); + } + } + + @Benchmark + @Threads(16) + public void jmap16Threads(Blackhole bh) { + int key = ThreadLocalRandom.current().nextInt(KEY_SPACE); + if (ThreadLocalRandom.current().nextInt(10) < 9) { + bh.consume(jmap.get("key" + key)); + } else { + jmap.put("key" + key, "val" + key); + } + } + + @Benchmark + @Threads(16) + public void chm16Threads(Blackhole bh) { + int key = ThreadLocalRandom.current().nextInt(KEY_SPACE); + if (ThreadLocalRandom.current().nextInt(10) < 9) { + bh.consume(chm.get("key" + key)); + } else { + chm.put("key" + key, "val" + key); + } + } +} \ No newline at end of file diff --git a/src/main/java/Jnode.java b/src/main/java/cache/Jnode.java similarity index 98% rename from src/main/java/Jnode.java rename to src/main/java/cache/Jnode.java index 6657532..40e1381 100644 --- a/src/main/java/Jnode.java +++ b/src/main/java/cache/Jnode.java @@ -1,3 +1,4 @@ +package cache; import java.nio.charset.StandardCharsets; import java.util.Objects; diff --git a/src/main/java/KVStore.java b/src/main/java/cache/KVStore.java similarity index 95% rename from src/main/java/KVStore.java rename to src/main/java/cache/KVStore.java index 3f8e967..782fa38 100644 --- a/src/main/java/KVStore.java +++ b/src/main/java/cache/KVStore.java @@ -1,3 +1,4 @@ +package cache; import java.io.BufferedReader; import java.io.BufferedWriter; import java.io.File; @@ -17,16 +18,16 @@ public class KVStore { // private static ConcurrentHashMap map = new // ConcurrentHashMap<>(16, 0.75f); private static Jmap map = new Jmap<>(0.75f); - private static TTLManager ttlManager; + private static TTLManager ttlManager; private static PrintWriter logWriter; private static final String LOG_FILE = "src/main/logs/logs.txt"; public static void main(String[] args) throws InterruptedException { - ttlManager = new TTLManager(map); - initializeLog(); - loadLog(); - interactiveMode(); - // stressTest(); + // ttlManager = new TTLManager(map); + // initializeLog(); + // loadLog(); + // interactiveMode(); + stressTest(); } private static void initializeLog() { @@ -173,7 +174,7 @@ public static void interactiveMode() { break; case "count": case "size": - System.out.println("Node count: " + Jmap.node_ct.get()); + System.out.println("Node count: " + map.node_ct.get()); break; case "help": System.out.println( @@ -200,9 +201,10 @@ public static void interactiveMode() { } public static void stressTest() throws InterruptedException { + System.out.println("running stress test... in kkvstore"); int writers = 10; - int readers = 40; - int operationsPerThread = 1000; + int readers = 10; + int operationsPerThread = 100; CountDownLatch startGate = new CountDownLatch(1); CountDownLatch endGate = new CountDownLatch(writers + readers); ExecutorService executor = Executors.newFixedThreadPool(writers + readers); @@ -247,6 +249,6 @@ public static void stressTest() throws InterruptedException { System.out.println("Test duration: " + (endTime - startTime) + "ms"); System.out.println("Expected nodes: " + (writers * operationsPerThread)); // System.out.println("actual nodes: " + map.size()); - System.out.println("Actual node_ct: " + Jmap.node_ct.get()); + System.out.println("Actual node_ct: " + map.node_ct.get()); } } diff --git a/src/main/java/cache/StressTest.java b/src/main/java/cache/StressTest.java new file mode 100644 index 0000000..5b74862 --- /dev/null +++ b/src/main/java/cache/StressTest.java @@ -0,0 +1,87 @@ +package cache; + +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; + +public class StressTest { + + static final int THREADS = 16; + static final int OPS_PER_THREAD = 50_000; + static final int KEY_SPACE = 2000; // will trigger multiple resizes + static final int TTL_MS = 100; + + public static void main(String[] args) throws InterruptedException { + System.out.println("Starting full stress test..."); + Jmap map = new Jmap<>(); + TTLManager ttlManager = new TTLManager<>(map); + + ExecutorService pool = Executors.newFixedThreadPool(THREADS); + CountDownLatch latch = new CountDownLatch(THREADS); + AtomicInteger errors = new AtomicInteger(0); + AtomicInteger corruptReads = new AtomicInteger(0); + + for (int t = 0; t < THREADS; t++) { + pool.submit(() -> { + try { + ThreadLocalRandom rng = ThreadLocalRandom.current(); + for (int i = 0; i < OPS_PER_THREAD; i++) { + int key = rng.nextInt(KEY_SPACE); + int op = rng.nextInt(10); + + if (op < 3) { + // 30% plain put + map.put("key" + key, "val" + key); + } else if (op < 5) { + // 20% putexp + long expiry = System.currentTimeMillis() + TTL_MS; + map.putexp("key" + key, "val" + key, TTL_MS); + ttlManager.schedule("key" + key, expiry); + } else if (op < 8) { + // 30% get with correctness check + String val = map.get("key" + key); + if (val != null && !val.equals("val" + key)) { + corruptReads.incrementAndGet(); + System.out.println("CORRUPTION: key" + key + " -> " + val); + } + } else if (op < 9) { + // 10% remove + map.remove("key" + key); + } else { + // 10% putexp with very short TTL to stress eviction + long expiry = System.currentTimeMillis() + 20; + map.putexp("key" + key, "val" + key, 20); + ttlManager.schedule("key" + key, expiry); + } + } + } catch (Exception e) { + errors.incrementAndGet(); + e.printStackTrace(); + } finally { + latch.countDown(); + } + }); + } + + boolean completed = latch.await(60, TimeUnit.SECONDS); + pool.shutdown(); + + // let TTL cleanup finish + Thread.sleep(TTL_MS * 3); + + System.out.println("--- Results ---"); + System.out.println("Completed: " + completed); + System.out.println("Exceptions: " + errors.get()); + System.out.println("Corrupt reads: " + corruptReads.get()); + System.out.println("Final node_ct: " + map.node_ct.get()); + + if (!completed) { + System.out.println("DEADLOCK — did not finish within 60 seconds"); + } else if (errors.get() > 0 || corruptReads.get() > 0) { + System.out.println("FAILED"); + } else if (map.node_ct.get() < 0) { + System.out.println("FAILED — negative node_ct"); + } else { + System.out.println("PASSED"); + } + } +} \ No newline at end of file diff --git a/src/main/java/TTLManager.java b/src/main/java/cache/TTLManager.java similarity index 94% rename from src/main/java/TTLManager.java rename to src/main/java/cache/TTLManager.java index dd1f13f..bdde0b1 100644 --- a/src/main/java/TTLManager.java +++ b/src/main/java/cache/TTLManager.java @@ -1,3 +1,4 @@ +package cache; import java.util.concurrent.DelayQueue; public class TTLManager { @@ -32,7 +33,7 @@ private void startCleanupThread() { if (currentExpiry != null && currentExpiry == entry.getExpiryTime()) { try { map.remove(entry.getKey()); - System.out.println("[TTL] Expired and removed key: " + entry.getKey()); + // System.out.println("[TTL] Expired and removed key: " + entry.getKey()); } catch (IllegalArgumentException e) { // Key was already removed, ignore } diff --git a/src/main/logs/logs.txt b/src/main/logs/logs.txt index b7dbbd1..40e6144 100644 --- a/src/main/logs/logs.txt +++ b/src/main/logs/logs.txt @@ -2,3 +2,4 @@ put 1 hello put we woo del we del 1 +put 1 10 diff --git a/src/test/java/TTLTest.java b/src/test/java/TTLTest.java index 81f308d..c271e19 100644 --- a/src/test/java/TTLTest.java +++ b/src/test/java/TTLTest.java @@ -1,6 +1,9 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import cache.Jmap; +import cache.TTLManager; + public class TTLTest { public static void main(String[] args) throws InterruptedException {