use indexedmatrixvalue and check with actions

Author: j143

src/main/java/org/apache/sysds/runtime/controlprogram/caching/OOCEvictionManager.java

@@ -78,7 +78,7 @@ public class OOCEvictionManager {
 	private static long _size;
 
 	// Cache structures: map key -> MatrixBlock and eviction deque (head=oldest block)
-	private static final Map<String, MatrixBlock> _cache = new HashMap<>();
+	private static final Map<String, IndexedMatrixValue> _cache = new HashMap<>();
 	private static final Deque<String> _evictDeque = new ArrayDeque<>();
 
 	// Single lock for synchronization
@@ -110,10 +110,10 @@ public static synchronized void put(long streamId, int blockId, IndexedMatrixVal
 		String key = streamId + "_" + blockId;
 
 		synchronized (lock) {
-			MatrixBlock old = _cache.remove(key);
+			IndexedMatrixValue old = _cache.remove(key); // remove old value
 			if (old != null) {
 				_evictDeque.remove(key);
-				_size -= estimateSerializedSize(old);
+				_size -= estimateSerializedSize((MatrixBlock) old.getValue());
 			}
 
 			try {
@@ -122,7 +122,7 @@ public static synchronized void put(long streamId, int blockId, IndexedMatrixVal
 				throw new DMLRuntimeException(e);
 			}
 
-			_cache.put(key, mb);
+			_cache.put(key, value); // put new value
 			_evictDeque.addLast(key); // add to end for FIFO/LRU
 			_size += size;
 		}
@@ -132,19 +132,19 @@ public static synchronized void put(long streamId, int blockId, IndexedMatrixVal
 	 * Get a block from the OOC cache (deserialize on read)
 	 */
 	public static synchronized IndexedMatrixValue get(long streamId, int blockId) {
+
 		String key = streamId + "_" + blockId;
-		MatrixBlock mb = (MatrixBlock) _cache.get(key);
+		IndexedMatrixValue imv = _cache.get(key);
 
 		synchronized (lock) {
-			if (mb != null && _policy == RPolicy.LRU) {
+			if (imv != null && _policy == RPolicy.LRU) {
 				_evictDeque.remove(key);
 				_evictDeque.addLast(key);
 			}
 		}
 
-		if (mb != null) {
-			MatrixIndexes ix = new MatrixIndexes(blockId + 1, 1);
-			return new IndexedMatrixValue(ix, mb);
+		if (imv != null) {
+			return imv;
 		} else {
 			try {
 				return loadFromDisk(streamId, blockId);
@@ -162,23 +162,23 @@ private static void evict(long requiredSize) throws IOException {
 		while(_size + requiredSize > _limit && !_evictDeque.isEmpty()) {
 			System.out.println("_size + requiredSize: " + _size +" + "+ requiredSize + "; _limit: " + _limit);
 			String oldKey = _evictDeque.removeLast();
-			MatrixBlock mbToEvict = (MatrixBlock) _cache.remove(oldKey);
+			IndexedMatrixValue toEvict = _cache.remove(oldKey);
+
+			if (toEvict == null) { continue;}
+			MatrixBlock mbToEvict = (MatrixBlock) toEvict.getValue();
 
-			if(mbToEvict != null) {
+			// Spill to disk
+			String filename = _spillDir + "/" + oldKey;
+			File spillDirFile = new File(_spillDir);
+			if (!spillDirFile.exists()) {
+				spillDirFile.mkdirs();
+			}
 
-				// Spill to disk
-				String filename = _spillDir + "/" + oldKey;
-				File spillDirFile = new File(_spillDir);
-				if (!spillDirFile.exists()) {
-					spillDirFile.mkdirs();
-				}
+			LocalFileUtils.writeMatrixBlockToLocal(filename, mbToEvict);
 
-				LocalFileUtils.writeMatrixBlockToLocal(filename, mbToEvict);
-				System.out.println("Evicting directory: "+ filename);
+			long freedSize = estimateSerializedSize(mbToEvict);
+			_size -= freedSize;
 
-				long freedSize = estimateSerializedSize(mbToEvict);
-				_size -= freedSize;
-			}
 		}
 	}