[SYSTEMDS-3891] [Major] Refactor Cache Manager / Performance Improvements / OOC Statistics

src/main/java/org/apache/sysds/api/DMLOptions.java

@@ -19,6 +19,10 @@
 
 package org.apache.sysds.api;
 
+import java.nio.file.Files;
+import java.nio.file.InvalidPathException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -66,6 +70,10 @@ public class DMLOptions {
 	public boolean              gpu           = false;            // Whether to use the GPU
 	public boolean              forceGPU      = false;            // Whether to ignore memory & estimates and always use the GPU
 	public boolean              ooc           = false;            // Whether to use the OOC backend
+	public boolean              oocLogEvents  = false;            // Whether to record I/O and task compute events (fine grained, may impact performance on many small tasks)
+	public String               oocLogPath    = "./";             // The directory where to save the recorded event logs (csv)
+	public boolean              oocStats      = false;            // Wether to record and print coarse grained ooc statistics
+	public int                  oocStatsCount = 10;               // Default ooc statistics count
 	public boolean              debug         = false;            // to go into debug mode to be able to step through a program
 	public String               filePath      = null;             // path to script
 	public String               script        = null;             // the script itself
@@ -105,7 +113,11 @@ public String toString() {
 			", fedStats=" + fedStats +
 			", fedStatsCount=" + fedStatsCount +
 			", fedMonitoring=" + fedMonitoring +
-			", fedMonitoringAddress" + fedMonitoringAddress +
+			", fedMonitoringAddress=" + fedMonitoringAddress +
+			", oocStats=" + oocStats +
+			", oocStatsCount=" + oocStatsCount +
+			", oocLogEvents=" + oocLogEvents +
+			", oocLogPath=" + oocLogPath +
 			", memStats=" + memStats +
 			", explainType=" + explainType +
 			", execMode=" + execMode +
@@ -193,7 +205,7 @@ else if (lineageType.equalsIgnoreCase("debugger"))
 			else if (execMode.equalsIgnoreCase("hybrid")) dmlOptions.execMode = ExecMode.HYBRID;
 			else if (execMode.equalsIgnoreCase("spark")) dmlOptions.execMode = ExecMode.SPARK;
 			else throw new org.apache.commons.cli.ParseException("Invalid argument specified for -exec option, must be one of [hadoop, singlenode, hybrid, HYBRID, spark]");
-	}
+		}
 		if (line.hasOption("explain")) {
 			dmlOptions.explainType = ExplainType.RUNTIME;
 			String explainType = line.getOptionValue("explain");
@@ -259,6 +271,33 @@ else if (lineageType.equalsIgnoreCase("debugger"))
 			}
 		}
 
+		dmlOptions.oocStats = line.hasOption("oocStats");
+		if (dmlOptions.oocStats) {
+			String oocStatsCount = line.getOptionValue("oocStats");
+			if (oocStatsCount != null) {
+				try {
+					dmlOptions.oocStatsCount = Integer.parseInt(oocStatsCount);
+				} catch (NumberFormatException e) {
+					throw new org.apache.commons.cli.ParseException("Invalid argument specified for -oocStats option, must be a valid integer");
+				}
+			}
+		}
+
+		dmlOptions.oocLogEvents = line.hasOption("oocLogEvents");
+		if (dmlOptions.oocLogEvents) {
+			String eventLogPath = line.getOptionValue("oocLogEvents");
+			if (eventLogPath != null) {
+				try {
+					Path p = Paths.get(eventLogPath);
+					if (!Files.isDirectory(p))
+						throw new org.apache.commons.cli.ParseException("Invalid argument specified for -oocLogEvents option, must be valid directory");
+				} catch (InvalidPathException e) {
+					throw new org.apache.commons.cli.ParseException("Invalid argument specified for -oocLogEvents option, must be a valid path");
+				}
+				dmlOptions.oocLogPath = eventLogPath;
+			}
+		}
+
 		dmlOptions.memStats = line.hasOption("mem");
 
 		dmlOptions.clean = line.hasOption("clean");
@@ -387,6 +426,12 @@ private static Options createCLIOptions() {
 		Option fedStatsOpt = OptionBuilder.withArgName("count")
 			.withDescription("monitors and reports summary execution statistics of federated workers; heavy hitter <count> is 10 unless overridden; default off")
 			.hasOptionalArg().create("fedStats");
+		Option oocStatsOpt = OptionBuilder
+			.withDescription("monitors and reports summary execution statistics of ooc operators and tasks; heavy hitter <count> is 10 unless overriden; default off")
+			.hasOptionalArg().create("oocStats");
+		Option oocLogEventsOpt = OptionBuilder
+			.withDescription("records fine grained events of compute tasks, I/O, and cache; -oocLogEvents [dir='./']")
+			.hasOptionalArg().create("oocLogEvents");
 		Option memOpt = OptionBuilder.withDescription("monitors and reports max memory consumption in CP; default off")
 			.create("mem");
 		Option explainOpt = OptionBuilder.withArgName("level")
@@ -452,6 +497,8 @@ private static Options createCLIOptions() {
 		options.addOption(statsOpt);
 		options.addOption(ngramsOpt);
 		options.addOption(fedStatsOpt);
+		options.addOption(oocStatsOpt);
+		options.addOption(oocLogEventsOpt);
 		options.addOption(memOpt);
 		options.addOption(explainOpt);
 		options.addOption(execOpt);

src/main/java/org/apache/sysds/api/DMLScript.java

@@ -75,6 +75,7 @@
 import org.apache.sysds.runtime.lineage.LineageCacheConfig;
 import org.apache.sysds.runtime.lineage.LineageCacheConfig.LineageCachePolicy;
 import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
+import org.apache.sysds.runtime.ooc.stats.OOCEventLog;
 import org.apache.sysds.runtime.util.CommonThreadPool;
 import org.apache.sysds.runtime.util.HDFSTool;
 import org.apache.sysds.runtime.util.LocalFileUtils;
@@ -149,6 +150,12 @@ public class DMLScript
 	public static boolean           SYNCHRONIZE_GPU      = true;
 	// Set OOC backend
 	public static boolean           USE_OOC              = DMLOptions.defaultOptions.ooc;
+	// Record and print OOC statistics
+	public static boolean           OOC_STATISTICS       = DMLOptions.defaultOptions.oocStats;
+	public static int               OOC_STATISTICS_COUNT = DMLOptions.defaultOptions.oocStatsCount;
+	// Record and save fine grained OOC event logs as csv to the specified dir
+	public static boolean           OOC_LOG_EVENTS       = DMLOptions.defaultOptions.oocLogEvents;
+	public static String            OOC_LOG_PATH         = DMLOptions.defaultOptions.oocLogPath;
 	// Enable eager CUDA free on rmvar
 	public static boolean           EAGER_CUDA_FREE      = false;
 
@@ -272,6 +279,10 @@ public static boolean executeScript( String[] args )
 			USE_ACCELERATOR       = dmlOptions.gpu;
 			FORCE_ACCELERATOR     = dmlOptions.forceGPU;
 			USE_OOC	              = dmlOptions.ooc;
+			OOC_STATISTICS        = dmlOptions.oocStats;
+			OOC_STATISTICS_COUNT  = dmlOptions.oocStatsCount;
+			OOC_LOG_EVENTS        = dmlOptions.oocLogEvents;
+			OOC_LOG_PATH          = dmlOptions.oocLogPath;
 			EXPLAIN               = dmlOptions.explainType;
 			EXEC_MODE             = dmlOptions.execMode;
 			LINEAGE               = dmlOptions.lineage;
@@ -323,6 +334,9 @@ public static boolean executeScript( String[] args )
 			LineageCacheConfig.setCachePolicy(LINEAGE_POLICY);
 			LineageCacheConfig.setEstimator(LINEAGE_ESTIMATE);
 
+			if (dmlOptions.oocLogEvents)
+				OOCEventLog.setup(100000);
+
 			String dmlScriptStr = readDMLScript(isFile, fileOrScript);
 			Map<String, String> argVals = dmlOptions.argVals;
 

src/main/java/org/apache/sysds/api/ScriptExecutorUtils.java

@@ -36,6 +36,7 @@
 import org.apache.sysds.runtime.instructions.gpu.context.GPUObject;
 import org.apache.sysds.runtime.lineage.LineageEstimatorStatistics;
 import org.apache.sysds.runtime.lineage.LineageGPUCacheEviction;
+import org.apache.sysds.runtime.ooc.cache.OOCCacheManager;
 import org.apache.sysds.utils.Statistics;
 
 public class ScriptExecutorUtils {
@@ -127,6 +128,9 @@ public static void executeRuntimeProgram(Program rtprog, ExecutionContext ec, DM
 
 			if (DMLScript.LINEAGE_ESTIMATE)
 				System.out.println(LineageEstimatorStatistics.displayLineageEstimates());
+
+			if (DMLScript.USE_OOC)
+				OOCCacheManager.reset();
 		}
 	}
 

src/main/java/org/apache/sysds/hops/rewrite/ProgramRewriter.java

@@ -69,15 +69,14 @@ public ProgramRewriter(boolean staticRewrites, boolean dynamicRewrites)
 
 		//initialize StatementBlock rewrite ruleSet (with fixed rewrite order)
 		_sbRuleSet = new ArrayList<>();
-		
-		
+
+
 		//STATIC REWRITES (which do not rely on size information)
 		if( staticRewrites )
 		{
 			//add static HOP DAG rewrite rules
 			_dagRuleSet.add(     new RewriteRemoveReadAfterWrite()               ); //dependency: before blocksize
 			_dagRuleSet.add(     new RewriteBlockSizeAndReblock()                );
-			_dagRuleSet.add(     new RewriteInjectOOCTee()                       );
 			if( OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION )
 				_dagRuleSet.add( new RewriteRemoveUnnecessaryCasts()             );
 			if( OptimizerUtils.ALLOW_COMMON_SUBEXPRESSION_ELIMINATION )
@@ -94,6 +93,7 @@ public ProgramRewriter(boolean staticRewrites, boolean dynamicRewrites)
 			if( OptimizerUtils.ALLOW_QUANTIZE_COMPRESS_REWRITE )
 				_dagRuleSet.add( new RewriteQuantizationFusedCompression()  	 );
 
+
 			//add statement block rewrite rules
 			if( OptimizerUtils.ALLOW_BRANCH_REMOVAL )
 				_sbRuleSet.add(  new RewriteRemoveUnnecessaryBranches()          ); //dependency: constant folding
@@ -152,6 +152,7 @@ public ProgramRewriter(boolean staticRewrites, boolean dynamicRewrites)
 			_dagRuleSet.add( new RewriteConstantFolding()                    ); //dependency: cse
 		_sbRuleSet.add(  new RewriteRemoveEmptyBasicBlocks()                 );
 		_sbRuleSet.add(  new RewriteRemoveEmptyForLoops()                    );
+		_sbRuleSet.add(      new RewriteInjectOOCTee()                       );
 	}
 
 	/**