[OOC] Add OOCPackedCache

src/main/java/org/apache/sysds/runtime/ooc/cache/OOCCache.java

@@ -24,10 +24,29 @@
 import java.util.function.LongUnaryOperator;
 
 public interface OOCCache {
+	/**
+	 * Pins an item backed by an allowance. A successful pin transfers memory ownership from the cache to the owner of
+	 * the allowance and guarantees data availability. While pinned, the bytes of the entry are not counted as
+	 * cache-owned memory.
+	 *
+	 * @param key
+	 * @param allowance
+	 * @return a non-null future of the pinned block entry; the future result is null if the required memory could not
+	 *         be reserved
+	 */
 	default OOCFuture<BlockEntry> pin(BlockKey key, MemoryAllowance allowance) {
 		return pin(key.getStreamId(), key.getSequenceNumber(), allowance);
 	}
 
+	/**
+	 * Pins an item backed by an allowance. If the allowance cannot reserve enough memory, this method will wait until
+	 * memory is available. A successful pin transfers memory ownership from the cache to the owner of the allowance and
+	 * guarantees data availability. While pinned, the bytes of the entry are not counted as cache-owned memory.
+	 *
+	 * @param key
+	 * @param allowance
+	 * @return a non-null future of the pinned block entry
+	 */
 	default OOCFuture<BlockEntry> pinAdmitted(BlockKey key, MemoryAllowance allowance) {
 		return pinAdmitted(key.getStreamId(), key.getSequenceNumber(), allowance);
 	}
@@ -59,9 +78,17 @@ default BlockEntry putPinned(BlockKey key, Object data, long size, MemoryAllowan
 	 */
 	OOCFuture<BlockEntry> pin(long sId, long tId, MemoryAllowance allowance);
 
-	default OOCFuture<BlockEntry> pinAdmitted(long sId, long tId, MemoryAllowance allowance) {
-		return pin(sId, tId, allowance);
-	}
+	/**
+	 * Pins an item backed by an allowance. If the allowance cannot reserve enough memory, this method will wait until
+	 * memory is available. A successful pin transfers memory ownership from the cache to the owner of the allowance and
+	 * guarantees data availability. While pinned, the bytes of the entry are not counted as cache-owned memory.
+	 *
+	 * @param sId
+	 * @param tId
+	 * @param allowance
+	 * @return a non-null future of the pinned block entry
+	 */
+	OOCFuture<BlockEntry> pinAdmitted(long sId, long tId, MemoryAllowance allowance);
 
 	/**
 	 * Pins an item backed by an allowance if it is already live in cache. A successful pin transfers memory ownership

src/main/java/org/apache/sysds/runtime/ooc/cache/io/SpillableObjectRegistry.java

@@ -20,13 +20,15 @@
 package org.apache.sysds.runtime.ooc.cache.io;
 
 import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue;
+import org.apache.sysds.runtime.ooc.cache.packed.PackedBlock;
 
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 
 public final class SpillableObjectRegistry {
 	private static final byte INDEXED_MATRIX_VALUE = 1;
+	private static final byte PACKED_BLOCK = 2;
 
 	private SpillableObjectRegistry() {
 	}
@@ -41,6 +43,7 @@ public static SpillableObject read(DataInput in) throws IOException {
 		byte type = in.readByte();
 		SpillableObject obj = switch(type) {
 			case INDEXED_MATRIX_VALUE -> new IndexedMatrixValue();
+			case PACKED_BLOCK -> new PackedBlock();
 			default -> throw new IOException("Unknown spillable object type: " + type);
 		};
 		obj.read(in);
@@ -50,6 +53,8 @@ public static SpillableObject read(DataInput in) throws IOException {
 	private static byte typeOf(SpillableObject obj) throws IOException {
 		if(obj instanceof IndexedMatrixValue)
 			return INDEXED_MATRIX_VALUE;
+		if(obj instanceof PackedBlock)
+			return PACKED_BLOCK;
 		throw new IOException("Unsupported spillable object type: " + obj.getClass().getName());
 	}
 }