You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by ar...@apache.org on 2022/01/21 12:11:29 UTC
[systemds] branch main updated: [SYSTEMDS-3185] Refactor Statistics module
This is an automated email from the ASF dual-hosted git repository.
arnabp20 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/systemds.git
The following commit(s) were added to refs/heads/main by this push:
new 1291aa9 [SYSTEMDS-3185] Refactor Statistics module
1291aa9 is described below
commit 1291aa9ee9794101973961d33da2d91415229dcb
Author: ywcb00 <yw...@ywcb.org>
AuthorDate: Fri Jan 21 13:11:04 2022 +0100
[SYSTEMDS-3185] Refactor Statistics module
This patch refactors the Statistics class by moving related
statistics into their own classes in a new package, utils/stats.
Moreover, this adds statistics for federated lookup table access
and splits the federated PUT request count into separate counts
for the individual object types.
Closes #1487
Closes #1519
---
src/main/java/org/apache/sysds/api/DMLScript.java | 96 +--
.../apache/sysds/hops/codegen/SpoofCompiler.java | 12 +-
.../codegen/opt/PlanSelectionFuseCostBased.java | 8 +-
.../codegen/opt/PlanSelectionFuseCostBasedV2.java | 16 +-
.../apache/sysds/runtime/codegen/CodegenUtils.java | 6 +-
.../runtime/controlprogram/BasicProgramBlock.java | 6 +-
.../runtime/controlprogram/ParForProgramBlock.java | 6 +-
.../sysds/runtime/controlprogram/ProgramBlock.java | 5 +-
.../context/SparkExecutionContext.java | 59 +-
.../controlprogram/federated/FederatedData.java | 1 +
.../federated/FederatedLookupTable.java | 15 +-
.../controlprogram/federated/FederatedRequest.java | 4 +-
.../federated/FederatedStatistics.java | 156 ++++-
.../paramserv/FederatedPSControlThread.java | 10 +-
.../controlprogram/paramserv/LocalPSWorker.java | 10 +-
.../controlprogram/paramserv/ParamServer.java | 16 +-
.../paramserv/SparkParamservUtils.java | 4 +-
.../parfor/opt/OptimizationWrapper.java | 6 +-
.../cp/ParamservBuiltinCPInstruction.java | 32 +-
.../instructions/cp/TriggerBroadcastTask.java | 4 +-
.../cp/TriggerRemoteOperationsTask.java | 7 +-
.../instructions/fed/InitFEDInstruction.java | 6 +-
.../runtime/matrix/data/LibMatrixDNNConv2d.java | 10 +-
.../sysds/runtime/matrix/data/LibMatrixNative.java | 32 +-
.../runtime/transform/encode/ColumnEncoder.java | 12 +-
.../runtime/transform/encode/ColumnEncoderBin.java | 10 +-
.../transform/encode/ColumnEncoderDummycode.java | 4 +-
.../transform/encode/ColumnEncoderFeatureHash.java | 4 +-
.../transform/encode/ColumnEncoderPassThrough.java | 4 +-
.../transform/encode/ColumnEncoderRecode.java | 10 +-
.../runtime/transform/encode/EncoderFactory.java | 8 +-
.../runtime/transform/encode/EncoderMVImpute.java | 6 +-
.../runtime/transform/encode/EncoderOmit.java | 4 +-
.../transform/encode/MultiColumnEncoder.java | 6 +-
.../java/org/apache/sysds/utils/Statistics.java | 704 ++-------------------
.../sysds/utils/stats/CodegenStatistics.java | 169 +++++
.../apache/sysds/utils/stats/NativeStatistics.java | 123 ++++
.../apache/sysds/utils/stats/ParForStatistics.java | 85 +++
.../sysds/utils/stats/ParamServStatistics.java | 171 +++++
.../sysds/utils/stats/RecompileStatistics.java | 75 +++
.../apache/sysds/utils/stats/SparkStatistics.java | 121 ++++
.../sysds/utils/stats/TransformStatistics.java | 175 +++++
src/test/config/SystemDS-MultiTenant-config.xml | 1 +
.../test/functions/async/AsyncBroadcastTest.java | 3 +-
.../test/functions/async/PrefetchRDDTest.java | 3 +-
.../test/functions/compress/LocalInstruction.java | 5 +-
.../multitenant/FederatedMultiTenantTest.java | 12 +-
.../functions/jmlc/JMLCParfor2ForCompileTest.java | 3 +-
.../functions/recompile/CSVReadInFunctionTest.java | 3 +-
49 files changed, 1369 insertions(+), 879 deletions(-)
diff --git a/src/main/java/org/apache/sysds/api/DMLScript.java b/src/main/java/org/apache/sysds/api/DMLScript.java
index ebb7f3d..4184db1 100644
--- a/src/main/java/org/apache/sysds/api/DMLScript.java
+++ b/src/main/java/org/apache/sysds/api/DMLScript.java
@@ -83,39 +83,65 @@ import org.apache.sysds.utils.Explain.ExplainType;
public class DMLScript
{
- private static ExecMode EXEC_MODE = DMLOptions.defaultOptions.execMode; // the execution mode
- public static boolean STATISTICS = DMLOptions.defaultOptions.stats; // whether to print statistics
- public static boolean JMLC_MEM_STATISTICS = false; // whether to gather memory use stats in JMLC
- public static int STATISTICS_COUNT = DMLOptions.defaultOptions.statsCount; // statistics maximum heavy hitter count
- public static int STATISTICS_MAX_WRAP_LEN = 30; // statistics maximum wrap length
- public static boolean FED_STATISTICS = DMLOptions.defaultOptions.fedStats; // whether to print federated statistics
- public static int FED_STATISTICS_COUNT = DMLOptions.defaultOptions.fedStatsCount; // federated statistics maximum heavy hitter count
- public static ExplainType EXPLAIN = DMLOptions.defaultOptions.explainType; // explain type
- public static String DML_FILE_PATH_ANTLR_PARSER = DMLOptions.defaultOptions.filePath; // filename of dml/pydml script
- public static String FLOATING_POINT_PRECISION = "double"; // data type to use internally
- public static boolean PRINT_GPU_MEMORY_INFO = false; // whether to print GPU memory-related information
- public static long EVICTION_SHADOW_BUFFER_MAX_BYTES = 0; // maximum number of bytes to use for shadow buffer
- public static long EVICTION_SHADOW_BUFFER_CURR_BYTES = 0; // number of bytes to use for shadow buffer
- public static double GPU_MEMORY_UTILIZATION_FACTOR = 0.9; // fraction of available GPU memory to use
- public static String GPU_MEMORY_ALLOCATOR = "cuda"; // GPU memory allocator to use
- public static boolean LINEAGE = DMLOptions.defaultOptions.lineage; // whether compute lineage trace
- public static boolean LINEAGE_DEDUP = DMLOptions.defaultOptions.lineage_dedup; // whether deduplicate lineage items
- public static ReuseCacheType LINEAGE_REUSE = DMLOptions.defaultOptions.linReuseType; // whether lineage-based reuse
- public static LineageCachePolicy LINEAGE_POLICY = DMLOptions.defaultOptions.linCachePolicy; // lineage cache eviction policy
- public static boolean LINEAGE_ESTIMATE = DMLOptions.defaultOptions.lineage_estimate; // whether estimate reuse benefits
- public static boolean LINEAGE_DEBUGGER = DMLOptions.defaultOptions.lineage_debugger; // whether enable lineage debugger
- public static boolean CHECK_PRIVACY = DMLOptions.defaultOptions.checkPrivacy; // Check which privacy constraints are loaded and checked during federated execution
-
- public static boolean USE_ACCELERATOR = DMLOptions.defaultOptions.gpu;
- public static boolean FORCE_ACCELERATOR = DMLOptions.defaultOptions.forceGPU;
- // whether to synchronize GPU after every instruction
- public static boolean SYNCHRONIZE_GPU = true;
- // whether to perform eager CUDA free on rmvar
- public static boolean EAGER_CUDA_FREE = false;
-
-
- public static boolean _suppressPrint2Stdout = false; // flag that indicates whether or not to suppress any prints to stdout
- public static boolean USE_LOCAL_SPARK_CONFIG = false; //set default local spark configuration - used for local testing
+ // Set the execution mode
+ private static ExecMode EXEC_MODE = DMLOptions.defaultOptions.execMode;
+ // Enable/disable to print statistics
+ public static boolean STATISTICS = DMLOptions.defaultOptions.stats;
+ // Enable/disable to gather memory use stats in JMLC
+ public static boolean JMLC_MEM_STATISTICS = false;
+ // Set maximum heavy hitter count
+ public static int STATISTICS_COUNT = DMLOptions.defaultOptions.statsCount;
+ // Set statistics maximum wrap length
+ public static int STATISTICS_MAX_WRAP_LEN = 30;
+ // Enable/disable to print federated statistics
+ public static boolean FED_STATISTICS = DMLOptions.defaultOptions.fedStats;
+ // Set federated statistics maximum heavy hitter count
+ public static int FED_STATISTICS_COUNT = DMLOptions.defaultOptions.fedStatsCount;
+ // Enable/disable this instance is a federated worker
+ public static boolean FED_WORKER = DMLOptions.defaultOptions.fedWorker;
+ // Set explain type
+ public static ExplainType EXPLAIN = DMLOptions.defaultOptions.explainType;
+ // Set filename of dml script
+ public static String DML_FILE_PATH_ANTLR_PARSER = DMLOptions.defaultOptions.filePath;
+ // Set data type to use internally
+ public static String FLOATING_POINT_PRECISION = "double";
+ // Enable/disable to print GPU memory-related information
+ public static boolean PRINT_GPU_MEMORY_INFO = false;
+ // Set maximum number of bytes to use for shadow buffer
+ public static long EVICTION_SHADOW_BUFFER_MAX_BYTES = 0;
+ // Set number of bytes to use for shadow buffer
+ public static long EVICTION_SHADOW_BUFFER_CURR_BYTES = 0;
+ // Set fraction of available GPU memory to use
+ public static double GPU_MEMORY_UTILIZATION_FACTOR = 0.9;
+ // Set GPU memory allocator to use
+ public static String GPU_MEMORY_ALLOCATOR = "cuda";
+ // Enable/disable lineage tracing
+ public static boolean LINEAGE = DMLOptions.defaultOptions.lineage;
+ // Enable/disable deduplicate lineage items
+ public static boolean LINEAGE_DEDUP = DMLOptions.defaultOptions.lineage_dedup;
+ // Enable/disable lineage-based reuse
+ public static ReuseCacheType LINEAGE_REUSE = DMLOptions.defaultOptions.linReuseType;
+ // Set lineage cache eviction policy
+ public static LineageCachePolicy LINEAGE_POLICY = DMLOptions.defaultOptions.linCachePolicy;
+ // Enable/disable estimate reuse benefits
+ public static boolean LINEAGE_ESTIMATE = DMLOptions.defaultOptions.lineage_estimate;
+ // Enable/disable lineage debugger
+ public static boolean LINEAGE_DEBUGGER = DMLOptions.defaultOptions.lineage_debugger;
+ // Check which privacy constraints are loaded and checked during federated execution
+ public static boolean CHECK_PRIVACY = DMLOptions.defaultOptions.checkPrivacy;
+ // Set accelerator
+ public static boolean USE_ACCELERATOR = DMLOptions.defaultOptions.gpu;
+ public static boolean FORCE_ACCELERATOR = DMLOptions.defaultOptions.forceGPU;
+ // Enable synchronizing GPU after every instruction
+ public static boolean SYNCHRONIZE_GPU = true;
+ // Enable eager CUDA free on rmvar
+ public static boolean EAGER_CUDA_FREE = false;
+
+
+ // flag that indicates whether or not to suppress any prints to stdout
+ public static boolean _suppressPrint2Stdout = false;
+ //set default local spark configuration - used for local testing
+ public static boolean USE_LOCAL_SPARK_CONFIG = false;
public static boolean _activeAM = false;
/**
* If true, allow DMLProgram to be generated while not halting due to validation errors/warnings
@@ -449,10 +475,8 @@ public class DMLScript
public static void setGlobalFlags(DMLConfig dmlconf) {
// Sets the GPUs to use for this process (a range, all GPUs, comma separated list or a specific GPU)
GPUContextPool.AVAILABLE_GPUS = dmlconf.getTextValue(DMLConfig.AVAILABLE_GPUS);
-
DMLScript.STATISTICS_MAX_WRAP_LEN = dmlconf.getIntValue(DMLConfig.STATS_MAX_WRAP_LEN);
NativeHelper.initialize(dmlconf.getTextValue(DMLConfig.NATIVE_BLAS_DIR), dmlconf.getTextValue(DMLConfig.NATIVE_BLAS).trim());
-
DMLScript.SYNCHRONIZE_GPU = dmlconf.getBooleanValue(DMLConfig.SYNCHRONIZE_GPU);
DMLScript.EAGER_CUDA_FREE = dmlconf.getBooleanValue(DMLConfig.EAGER_CUDA_FREE);
DMLScript.PRINT_GPU_MEMORY_INFO = dmlconf.getBooleanValue(DMLConfig.PRINT_GPU_MEMORY_INFO);
@@ -461,7 +485,7 @@ public class DMLScript
if(DMLScript.GPU_MEMORY_UTILIZATION_FACTOR < 0) {
throw new RuntimeException("Incorrect value (" + DMLScript.GPU_MEMORY_UTILIZATION_FACTOR + ") for the configuration:" + DMLConfig.GPU_MEMORY_UTILIZATION_FACTOR);
}
-
+ DMLScript.USE_LOCAL_SPARK_CONFIG |= dmlconf.getBooleanValue(DMLConfig.USE_LOCAL_SPARK_CONFIG);
DMLScript.FLOATING_POINT_PRECISION = dmlconf.getTextValue(DMLConfig.FLOATING_POINT_PRECISION);
org.apache.sysds.runtime.matrix.data.LibMatrixCUDA.resetFloatingPointPrecision();
if(DMLScript.FLOATING_POINT_PRECISION.equals("double")) {
diff --git a/src/main/java/org/apache/sysds/hops/codegen/SpoofCompiler.java b/src/main/java/org/apache/sysds/hops/codegen/SpoofCompiler.java
index 4886691..ade8877 100644
--- a/src/main/java/org/apache/sysds/hops/codegen/SpoofCompiler.java
+++ b/src/main/java/org/apache/sysds/hops/codegen/SpoofCompiler.java
@@ -96,7 +96,7 @@ import org.apache.sysds.runtime.lineage.LineageItemUtils;
import org.apache.sysds.runtime.matrix.data.Pair;
import org.apache.sysds.utils.Explain;
import org.apache.sysds.utils.NativeHelper;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.CodegenStatistics;
import java.io.File;
import java.io.IOException;
@@ -563,7 +563,7 @@ public class SpoofCompiler {
}
else {
if( DMLScript.STATISTICS )
- Statistics.incrementCodegenOpCacheHits();
+ CodegenStatistics.incrementOpCacheHits();
if(CodegenUtils.getCUDAopID(cla.getName()) != null) {
tmp.getValue().setGeneratorAPI(GeneratorAPI.CUDA);
tmp.getValue().setVarName(cla.getName().split("\\.")[1]);
@@ -586,7 +586,7 @@ public class SpoofCompiler {
clas.put(cplan.getKey(), new Pair<Hop[], Class<?>>(tmp.getKey(), cla));
}
if( DMLScript.STATISTICS )
- Statistics.incrementCodegenOpCacheTotal();
+ CodegenStatistics.incrementOpCacheTotal();
}
//create modified hop dag (operator replacement and CSE)
@@ -612,8 +612,8 @@ public class SpoofCompiler {
}
if( DMLScript.STATISTICS ) {
- Statistics.incrementCodegenDAGCompile();
- Statistics.incrementCodegenCompileTime(System.nanoTime()-t0);
+ CodegenStatistics.incrementDAGCompile();
+ CodegenStatistics.incrementCompileTime(System.nanoTime()-t0);
}
Hop.resetVisitStatus(roots);
@@ -738,7 +738,7 @@ public class SpoofCompiler {
if( tmp != null ) {
cplans.put(hop.getHopID(), tmp);
if (DMLScript.STATISTICS)
- Statistics.incrementCodegenCPlanCompile(1);
+ CodegenStatistics.incrementCPlanCompile(1);
}
}
diff --git a/src/main/java/org/apache/sysds/hops/codegen/opt/PlanSelectionFuseCostBased.java b/src/main/java/org/apache/sysds/hops/codegen/opt/PlanSelectionFuseCostBased.java
index e19b5a5..6d40895 100644
--- a/src/main/java/org/apache/sysds/hops/codegen/opt/PlanSelectionFuseCostBased.java
+++ b/src/main/java/org/apache/sysds/hops/codegen/opt/PlanSelectionFuseCostBased.java
@@ -58,7 +58,7 @@ import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyze
import org.apache.sysds.runtime.controlprogram.parfor.util.IDSequence;
import org.apache.sysds.runtime.util.CollectionUtils;
import org.apache.sysds.runtime.util.UtilFunctions;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.CodegenStatistics;
/**
* This cost-based plan selection algorithm chooses fused operators
@@ -107,7 +107,7 @@ public class PlanSelectionFuseCostBased extends PlanSelection
//maintain statistics
if( DMLScript.STATISTICS )
- Statistics.incrementCodegenEnumAll(UtilFunctions.pow(2, sumMatPoints));
+ CodegenStatistics.incrementEnumAll(UtilFunctions.pow(2, sumMatPoints));
}
//within-partition multi-agg templates
@@ -396,8 +396,8 @@ public class PlanSelectionFuseCostBased extends PlanSelection
}
if( DMLScript.STATISTICS ) {
- Statistics.incrementCodegenEnumAllP(len);
- Statistics.incrementCodegenEnumEval(len);
+ CodegenStatistics.incrementEnumAllP(len);
+ CodegenStatistics.incrementEnumEval(len);
}
//prune memo table wrt best plan and select plans
diff --git a/src/main/java/org/apache/sysds/hops/codegen/opt/PlanSelectionFuseCostBasedV2.java b/src/main/java/org/apache/sysds/hops/codegen/opt/PlanSelectionFuseCostBasedV2.java
index c6cfe9f..f077134 100644
--- a/src/main/java/org/apache/sysds/hops/codegen/opt/PlanSelectionFuseCostBasedV2.java
+++ b/src/main/java/org/apache/sysds/hops/codegen/opt/PlanSelectionFuseCostBasedV2.java
@@ -63,7 +63,7 @@ import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyze
import org.apache.sysds.runtime.controlprogram.parfor.util.IDSequence;
import org.apache.sysds.runtime.util.CollectionUtils;
import org.apache.sysds.runtime.util.UtilFunctions;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.CodegenStatistics;
/**
* This cost-based plan selection algorithm chooses fused operators
@@ -148,7 +148,7 @@ public class PlanSelectionFuseCostBasedV2 extends PlanSelection
if( sumMatPoints >= 63 )
LOG.warn("Long overflow on maintaining codegen statistics "
+ "for a DAG with "+sumMatPoints+" interesting points.");
- Statistics.incrementCodegenEnumAll(UtilFunctions.pow(2, sumMatPoints));
+ CodegenStatistics.incrementEnumAll(UtilFunctions.pow(2, sumMatPoints));
}
}
@@ -246,7 +246,7 @@ public class PlanSelectionFuseCostBasedV2 extends PlanSelection
pKey = new PartitionSignature(part, matPoints.length, costs, C0, CN);
boolean[] plan = getPlan(pKey);
if( plan != null ) {
- Statistics.incrementCodegenEnumAllP((rgraph!=null||!STRUCTURAL_PRUNING)?len:0);
+ CodegenStatistics.incrementEnumAllP((rgraph!=null||!STRUCTURAL_PRUNING)?len:0);
return plan;
}
}
@@ -314,9 +314,9 @@ public class PlanSelectionFuseCostBasedV2 extends PlanSelection
}
if( DMLScript.STATISTICS ) {
- Statistics.incrementCodegenEnumAllP((rgraph!=null||!STRUCTURAL_PRUNING)?len:0);
- Statistics.incrementCodegenEnumEval(numEvalPlans);
- Statistics.incrementCodegenEnumEvalP(numEvalPartPlans);
+ CodegenStatistics.incrementEnumAllP((rgraph!=null||!STRUCTURAL_PRUNING)?len:0);
+ CodegenStatistics.incrementEnumEval(numEvalPlans);
+ CodegenStatistics.incrementEnumEvalP(numEvalPartPlans);
}
if( LOG.isTraceEnabled() )
LOG.trace("Enum: Optimal plan: "+Arrays.toString(bestPlan));
@@ -1025,8 +1025,8 @@ public class PlanSelectionFuseCostBasedV2 extends PlanSelection
}
if( DMLScript.STATISTICS ) {
if( plan != null )
- Statistics.incrementCodegenPlanCacheHits();
- Statistics.incrementCodegenPlanCacheTotal();
+ CodegenStatistics.incrementPlanCacheHits();
+ CodegenStatistics.incrementPlanCacheTotal();
}
return plan;
}
diff --git a/src/main/java/org/apache/sysds/runtime/codegen/CodegenUtils.java b/src/main/java/org/apache/sysds/runtime/codegen/CodegenUtils.java
index b2b896b..d5bbdfa 100644
--- a/src/main/java/org/apache/sysds/runtime/codegen/CodegenUtils.java
+++ b/src/main/java/org/apache/sysds/runtime/codegen/CodegenUtils.java
@@ -31,7 +31,7 @@ import org.apache.sysds.runtime.codegen.SpoofOperator.SideInputSparseCell;
import org.apache.sysds.runtime.io.IOUtilFunctions;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.LocalFileUtils;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.CodegenStatistics;
import org.codehaus.janino.SimpleCompiler;
import javax.tools.Diagnostic;
@@ -88,8 +88,8 @@ public class CodegenUtils
_cache.put(name, ret);
if( DMLScript.STATISTICS ) {
- Statistics.incrementCodegenClassCompile();
- Statistics.incrementCodegenClassCompileTime(System.nanoTime()-t0);
+ CodegenStatistics.incrementClassCompile();
+ CodegenStatistics.incrementClassCompileTime(System.nanoTime()-t0);
}
return ret;
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/BasicProgramBlock.java b/src/main/java/org/apache/sysds/runtime/controlprogram/BasicProgramBlock.java
index fd95f38..1686055 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/BasicProgramBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/BasicProgramBlock.java
@@ -33,7 +33,7 @@ import org.apache.sysds.runtime.lineage.LineageCacheConfig;
import org.apache.sysds.runtime.lineage.LineageCacheStatistics;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.lineage.LineageItemUtils;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.RecompileStatistics;
public class BasicProgramBlock extends ProgramBlock
{
@@ -96,9 +96,9 @@ public class BasicProgramBlock extends ProgramBlock
}
if( DMLScript.STATISTICS ){
long t1 = System.nanoTime();
- Statistics.incrementHOPRecompileTime(t1-t0);
+ RecompileStatistics.incrementRecompileTime(t1-t0);
if( tmp!=_inst )
- Statistics.incrementHOPRecompileSB();
+ RecompileStatistics.incrementRecompileSB();
}
}
catch(Exception ex)
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java b/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java
index f7ead3c..dc60a79 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java
@@ -90,7 +90,7 @@ import org.apache.sysds.runtime.meta.DataCharacteristics;
import org.apache.sysds.runtime.util.CollectionUtils;
import org.apache.sysds.runtime.util.ProgramConverter;
import org.apache.sysds.runtime.util.UtilFunctions;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.ParForStatistics;
import java.io.Serializable;
import java.util.ArrayList;
@@ -760,7 +760,7 @@ public class ParForProgramBlock extends ForProgramBlock
//maintain statistics
long tinit = (long) time.stop();
if( DMLScript.STATISTICS )
- Statistics.incrementParForInitTime(tinit);
+ ParForStatistics.incrementInitTime(tinit);
if( _monitor )
StatisticMonitor.putPFStat(_ID, Stat.PARFOR_INIT_PARWRK_T, tinit);
@@ -1515,7 +1515,7 @@ public class ParForProgramBlock extends ForProgramBlock
throw new DMLRuntimeException("PARFOR: Number of executed tasks does not match the number of created tasks: tasks "+numTasks+"/"+expTasks+", iters "+numIters+"/"+expIters+".");
if( DMLScript.STATISTICS )
- Statistics.incrementParForMergeTime((long) time.stop());
+ ParForStatistics.incrementMergeTime((long) time.stop());
}
/**
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/ProgramBlock.java b/src/main/java/org/apache/sysds/runtime/controlprogram/ProgramBlock.java
index bad2ab7..75fe42d 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/ProgramBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/ProgramBlock.java
@@ -54,6 +54,7 @@ import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.meta.MetaData;
import org.apache.sysds.runtime.meta.MetaDataFormat;
import org.apache.sysds.runtime.privacy.propagation.PrivacyPropagator;
+import org.apache.sysds.utils.stats.RecompileStatistics;
import org.apache.sysds.utils.Statistics;
public abstract class ProgramBlock implements ParseInfo {
@@ -168,9 +169,9 @@ public abstract class ProgramBlock implements ParseInfo {
}
if(DMLScript.STATISTICS) {
long t1 = System.nanoTime();
- Statistics.incrementHOPRecompileTime(t1 - t0);
+ RecompileStatistics.incrementRecompileTime(t1 - t0);
if(tmp != inst)
- Statistics.incrementHOPRecompilePred();
+ RecompileStatistics.incrementRecompilePred();
}
}
catch(Exception ex) {
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/context/SparkExecutionContext.java b/src/main/java/org/apache/sysds/runtime/controlprogram/context/SparkExecutionContext.java
index af96a21..9ab149d 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/context/SparkExecutionContext.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/context/SparkExecutionContext.java
@@ -98,6 +98,7 @@ import org.apache.sysds.runtime.util.HDFSTool;
import org.apache.sysds.runtime.util.UtilFunctions;
import org.apache.sysds.utils.MLContextProxy;
import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.SparkStatistics;
import scala.Tuple2;
@@ -220,15 +221,10 @@ public class SparkExecutionContext extends ExecutionContext
else
{
final SparkConf conf = createSystemDSSparkConf();
- final DMLConfig dmlConfig= ConfigurationManager.getDMLConfig();
- // Use Spark local config, if already set to True ... keep true, otherwise look up if it should be local.
- DMLScript.USE_LOCAL_SPARK_CONFIG = DMLScript.USE_LOCAL_SPARK_CONFIG ? true : dmlConfig.getBooleanValue(DMLConfig.USE_LOCAL_SPARK_CONFIG);
-
if(DMLScript.USE_LOCAL_SPARK_CONFIG)
setLocalConfSettings(conf);
_spctx = createContext(conf);
-
if(DMLScript.USE_LOCAL_SPARK_CONFIG)
_spctx.setCheckpointDir("/tmp/systemds_spark_cache_" + DMLScript.getUUID());
@@ -250,7 +246,7 @@ public class SparkExecutionContext extends ExecutionContext
//statistics maintenance
if( DMLScript.STATISTICS ){
- Statistics.setSparkCtxCreateTime(System.nanoTime()-t0);
+ SparkStatistics.setCtxCreateTime(System.nanoTime()-t0);
}
}
@@ -655,8 +651,8 @@ public class SparkExecutionContext extends ExecutionContext
CacheableData.addBroadcastSize(cd.getBroadcastHandle().getSize());
if (DMLScript.STATISTICS) {
- Statistics.accSparkBroadCastTime(System.nanoTime() - t0);
- Statistics.incSparkBroadcastCount(1);
+ SparkStatistics.accBroadCastTime(System.nanoTime() - t0);
+ SparkStatistics.incBroadcastCount(1);
}
}
}
@@ -721,8 +717,8 @@ public class SparkExecutionContext extends ExecutionContext
CacheableData.addBroadcastSize(mo.getBroadcastHandle().getSize());
if (DMLScript.STATISTICS) {
- Statistics.accSparkBroadCastTime(System.nanoTime() - t0);
- Statistics.incSparkBroadcastCount(1);
+ SparkStatistics.accBroadCastTime(System.nanoTime() - t0);
+ SparkStatistics.incBroadcastCount(1);
}
}
}
@@ -782,8 +778,8 @@ public class SparkExecutionContext extends ExecutionContext
CacheableData.addBroadcastSize(to.getBroadcastHandle().getSize());
if (DMLScript.STATISTICS) {
- Statistics.accSparkBroadCastTime(System.nanoTime() - t0);
- Statistics.incSparkBroadcastCount(1);
+ SparkStatistics.accBroadCastTime(System.nanoTime() - t0);
+ SparkStatistics.incBroadcastCount(1);
}
}
return bret;
@@ -848,8 +844,8 @@ public class SparkExecutionContext extends ExecutionContext
CacheableData.addBroadcastSize(fo.getBroadcastHandle().getSize());
if (DMLScript.STATISTICS) {
- Statistics.accSparkBroadCastTime(System.nanoTime() - t0);
- Statistics.incSparkBroadcastCount(1);
+ SparkStatistics.accBroadCastTime(System.nanoTime() - t0);
+ SparkStatistics.incBroadcastCount(1);
}
}
return bret;
@@ -904,8 +900,8 @@ public class SparkExecutionContext extends ExecutionContext
sc.parallelizePairs(list, numParts) : sc.parallelizePairs(list);
if (DMLScript.STATISTICS) {
- Statistics.accSparkParallelizeTime(System.nanoTime() - t0);
- Statistics.incSparkParallelizeCount(1);
+ SparkStatistics.accParallelizeTime(System.nanoTime() - t0);
+ SparkStatistics.incParallelizeCount(1);
}
return result;
@@ -946,8 +942,8 @@ public class SparkExecutionContext extends ExecutionContext
sc.parallelizePairs(list, numParts) : sc.parallelizePairs(list);
if (DMLScript.STATISTICS) {
- Statistics.accSparkParallelizeTime(System.nanoTime() - t0);
- Statistics.incSparkParallelizeCount(1);
+ SparkStatistics.accParallelizeTime(System.nanoTime() - t0);
+ SparkStatistics.incParallelizeCount(1);
}
return result;
@@ -1015,8 +1011,8 @@ public class SparkExecutionContext extends ExecutionContext
JavaPairRDD<Long,FrameBlock> result = sc.parallelizePairs(list);
if (DMLScript.STATISTICS) {
- Statistics.accSparkParallelizeTime(System.nanoTime() - t0);
- Statistics.incSparkParallelizeCount(1);
+ SparkStatistics.accParallelizeTime(System.nanoTime() - t0);
+ SparkStatistics.incParallelizeCount(1);
}
return result;
@@ -1058,9 +1054,9 @@ public class SparkExecutionContext extends ExecutionContext
final Boolean singleBlock = rlen <= blen && clen <= blen ;
final MatrixBlock out = singleBlock ? toMatrixBlockSingleBlock(rdd, rlen, clen) : toMatrixBlockMultiBlock(rdd, rlen, clen, blen, nnz);
- if (DMLScript.STATISTICS)
- Statistics.accSparkCollectTime(System.nanoTime() - t0);
-
+ if (DMLScript.STATISTICS)
+ SparkStatistics.accCollectTime(System.nanoTime() - t0);
+
return out;
}
@@ -1242,9 +1238,8 @@ public class SparkExecutionContext extends ExecutionContext
out.recomputeNonZeros();
out.examSparsity();
- if (DMLScript.STATISTICS)
- Statistics.accSparkCollectTime(System.nanoTime() - t0);
-
+ if (DMLScript.STATISTICS)
+ SparkStatistics.accCollectTime(System.nanoTime() - t0);
return out;
}
@@ -1288,8 +1283,8 @@ public class SparkExecutionContext extends ExecutionContext
// TODO post-processing output tensor (nnz, sparsity)
- if (DMLScript.STATISTICS)
- Statistics.accSparkCollectTime(System.nanoTime() - t0);
+ if (DMLScript.STATISTICS)
+ SparkStatistics.accCollectTime(System.nanoTime() - t0);
return out;
}
@@ -1309,8 +1304,8 @@ public class SparkExecutionContext extends ExecutionContext
out.setBlock((int)ix.getRowIndex(), (int)ix.getColumnIndex(), block);
}
- if (DMLScript.STATISTICS)
- Statistics.accSparkCollectTime(System.nanoTime() - t0);
+ if (DMLScript.STATISTICS)
+ SparkStatistics.accCollectTime(System.nanoTime() - t0);
return out;
}
@@ -1348,8 +1343,8 @@ public class SparkExecutionContext extends ExecutionContext
}
}
- if (DMLScript.STATISTICS)
- Statistics.accSparkCollectTime(System.nanoTime() - t0);
+ if (DMLScript.STATISTICS)
+ SparkStatistics.accCollectTime(System.nanoTime() - t0);
return out;
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java
index 6f063fb..07079b7 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java
@@ -33,6 +33,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.sysds.common.Types;
import org.apache.sysds.conf.ConfigurationManager;
import org.apache.sysds.conf.DMLConfig;
+
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest.RequestType;
import org.apache.sysds.runtime.meta.MetaData;
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedLookupTable.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedLookupTable.java
index 905ab57..c3fc1f5 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedLookupTable.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedLookupTable.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.Objects;
import org.apache.log4j.Logger;
+import org.apache.sysds.api.DMLScript;
/**
* Lookup table mapping from a FedUniqueCoordID (funCID) to an
@@ -57,15 +58,20 @@ public class FederatedLookupTable {
*/
public ExecutionContextMap getECM(String host, long pid) {
log.trace("Getting the ExecutionContextMap for coordinator " + pid + "@" + host);
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
FedUniqueCoordID funCID = new FedUniqueCoordID(host, pid);
ExecutionContextMap ecm = _lookup_table.computeIfAbsent(funCID,
- k -> new ExecutionContextMap());
+ k -> createNewECM());
if(ecm == null) {
log.error("Computing federated execution context map failed. "
+ "No valid resolution for " + funCID.toString() + " found.");
throw new FederatedWorkerHandlerException("Computing federated execution context map failed. "
+ "No valid resolution for " + funCID.toString() + " found.");
}
+ if(DMLScript.STATISTICS) {
+ FederatedStatistics.incFedLookupTableGetCount();
+ FederatedStatistics.incFedLookupTableGetTime(System.nanoTime() - t0);
+ }
return ecm;
}
@@ -82,6 +88,13 @@ public class FederatedLookupTable {
return _lookup_table.containsKey(funCID);
}
+ private ExecutionContextMap createNewECM() {
+ ExecutionContextMap ecm = new ExecutionContextMap();
+ if(DMLScript.STATISTICS)
+ FederatedStatistics.incFedLookupTableEntryCount();
+ return ecm;
+ }
+
@Override
public String toString() {
return _lookup_table.toString();
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedRequest.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedRequest.java
index ad9a711..9e38527 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedRequest.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedRequest.java
@@ -35,7 +35,6 @@ import org.apache.sysds.runtime.controlprogram.caching.CacheDataOutput;
import org.apache.sysds.runtime.controlprogram.caching.LazyWriteBuffer;
import org.apache.sysds.runtime.controlprogram.parfor.util.IDHandler;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
-import org.apache.sysds.utils.Statistics;
public class FederatedRequest implements Serializable {
private static final long serialVersionUID = 5946781306963870394L;
@@ -72,7 +71,8 @@ public class FederatedRequest implements Serializable {
}
public FederatedRequest(RequestType method, long id, List<Object> data) {
- Statistics.incFederated(method);
+ if(DMLScript.STATISTICS)
+ FederatedStatistics.incFederated(method, data);
_method = method;
_id = id;
_data = data;
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedStatistics.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedStatistics.java
index 791582f..b5e69b0 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedStatistics.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedStatistics.java
@@ -30,6 +30,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;
+import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.Future;
import javax.net.ssl.SSLException;
@@ -43,11 +44,126 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedStatistics.Fed
import org.apache.sysds.runtime.controlprogram.federated.FederatedStatistics.FedStatsCollection.GCStatsCollection;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.instructions.cp.Data;
+import org.apache.sysds.runtime.instructions.cp.ListObject;
+import org.apache.sysds.runtime.instructions.cp.ScalarObject;
import org.apache.sysds.runtime.lineage.LineageItem;
+import org.apache.sysds.runtime.matrix.data.FrameBlock;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.meta.MatrixCharacteristics;
import org.apache.sysds.utils.Statistics;
public class FederatedStatistics {
+ // stats of the federated worker on the coordinator site
private static Set<Pair<String, Integer>> _fedWorkerAddresses = new HashSet<>();
+ private static final LongAdder readCount = new LongAdder();
+ private static final LongAdder putScalarCount = new LongAdder();
+ private static final LongAdder putListCount = new LongAdder();
+ private static final LongAdder putMatrixCount = new LongAdder();
+ private static final LongAdder putFrameCount = new LongAdder();
+ private static final LongAdder putMatCharCount = new LongAdder();
+ private static final LongAdder putMatrixBytes = new LongAdder();
+ private static final LongAdder putFrameBytes = new LongAdder();
+ private static final LongAdder getCount = new LongAdder();
+ private static final LongAdder executeInstructionCount = new LongAdder();
+ private static final LongAdder executeUDFCount = new LongAdder();
+ private static final LongAdder asyncPrefetchCount = new LongAdder();
+
+ // stats on the federated worker itself
+ private static final LongAdder fedLookupTableGetCount = new LongAdder();
+ private static final LongAdder fedLookupTableGetTime = new LongAdder(); // in milli sec
+ private static final LongAdder fedLookupTableEntryCount = new LongAdder();
+
+ public static synchronized void incFederated(RequestType rqt, List<Object> data){
+ switch (rqt) {
+ case READ_VAR:
+ readCount.increment();
+ break;
+ case PUT_VAR:
+ if(data.get(0) instanceof MatrixBlock) {
+ putMatrixCount.increment();
+ putMatrixBytes.add(((MatrixBlock)data.get(0)).getInMemorySize());
+ }
+ else if(data.get(0) instanceof FrameBlock) {
+ putFrameCount.increment();
+ putFrameBytes.add(((FrameBlock)data.get(0)).getInMemorySize());
+ }
+ else if(data.get(0) instanceof ScalarObject)
+ putScalarCount.increment();
+ else if(data.get(0) instanceof ListObject)
+ putListCount.increment();
+ else if(data.get(0) instanceof MatrixCharacteristics)
+ putMatCharCount.increment();
+ break;
+ case GET_VAR:
+ getCount.increment();
+ break;
+ case EXEC_INST:
+ executeInstructionCount.increment();
+ break;
+ case EXEC_UDF:
+ executeUDFCount.increment();
+ break;
+ default:
+ break;
+ }
+ }
+
+ public static void incAsyncPrefetchCount(long c) {
+ asyncPrefetchCount.add(c);
+ }
+
+ public static long getTotalPutCount() {
+ return putScalarCount.longValue() + putListCount.longValue()
+ + putMatrixCount.longValue() + putFrameCount.longValue()
+ + putMatCharCount.longValue();
+ }
+
+ public static void reset() {
+ readCount.reset();
+ putScalarCount.reset();
+ putListCount.reset();
+ putMatrixCount.reset();
+ putFrameCount.reset();
+ putMatCharCount.reset();
+ putMatrixBytes.reset();
+ putFrameBytes.reset();
+ getCount.reset();
+ executeInstructionCount.reset();
+ executeUDFCount.reset();
+ asyncPrefetchCount.reset();
+ fedLookupTableGetCount.reset();
+ fedLookupTableGetTime.reset();
+ fedLookupTableEntryCount.reset();
+ }
+
+ public static String displayFedIOExecStatistics() {
+ if( readCount.longValue() > 0){ // only if there happened something on the federated worker
+ StringBuilder sb = new StringBuilder();
+ sb.append("Federated I/O (Read, Put, Get):\t" +
+ readCount.longValue() + "/" +
+ getTotalPutCount() + "/" +
+ getCount.longValue() + ".\n");
+ if(getTotalPutCount() > 0)
+ sb.append("Fed Put (Sca/Lis/Mat/Fra/MC):\t" +
+ putScalarCount.longValue() + "/" +
+ putListCount.longValue() + "/" +
+ putMatrixCount.longValue() + "/" +
+ putFrameCount.longValue() + "/" +
+ putMatCharCount.longValue() + ".\n");
+ if(putMatrixBytes.longValue() > 0 || putFrameBytes.longValue() > 0)
+ sb.append("Fed Put Bytes (Mat/Frame):\t" +
+ putMatrixBytes.longValue() + "/" +
+ putFrameBytes.longValue() + " Bytes.\n");
+ sb.append("Federated Execute (Inst, UDF):\t" +
+ executeInstructionCount.longValue() + "/" +
+ executeUDFCount.longValue() + ".\n");
+ sb.append("Federated prefetch count:\t" +
+ asyncPrefetchCount.longValue() + ".\n");
+ return sb.toString();
+ }
+ return "";
+ }
+
public static void registerFedWorker(String host, int port) {
_fedWorkerAddresses.add(new ImmutablePair<>(host, Integer.valueOf(port)));
@@ -63,7 +179,7 @@ public class FederatedStatistics {
return sb.toString();
}
- public static String displayFedStatistics(int numHeavyHitters) {
+ public static String displayStatistics(int numHeavyHitters) {
StringBuilder sb = new StringBuilder();
FedStatsCollection fedStats = collectFedStats();
sb.append("SystemDS Federated Statistics:\n");
@@ -74,7 +190,7 @@ public class FederatedStatistics {
return sb.toString();
}
- public static String displayCacheStats(CacheStatsCollection csc) {
+ private static String displayCacheStats(CacheStatsCollection csc) {
StringBuilder sb = new StringBuilder();
sb.append(String.format("Cache hits (Mem/Li/WB/FS/HDFS):\t%d/%d/%d/%d/%d.\n",
csc.memHits, csc.linHits, csc.fsBuffHits, csc.fsHits, csc.hdfsHits));
@@ -85,18 +201,19 @@ public class FederatedStatistics {
return sb.toString();
}
- public static String displayGCStats(GCStatsCollection gcsc) {
+ private static String displayGCStats(GCStatsCollection gcsc) {
StringBuilder sb = new StringBuilder();
sb.append(String.format("Total JVM GC count:\t\t%d.\n", gcsc.gcCount));
sb.append(String.format("Total JVM GC time:\t\t%.3f sec.\n", gcsc.gcTime));
return sb.toString();
}
- public static String displayHeavyHitters(HashMap<String, Pair<Long, Double>> heavyHitters) {
+ @SuppressWarnings("unused")
+ private static String displayHeavyHitters(HashMap<String, Pair<Long, Double>> heavyHitters) {
return displayHeavyHitters(heavyHitters, 10);
}
- public static String displayHeavyHitters(HashMap<String, Pair<Long, Double>> heavyHitters, int num) {
+ private static String displayHeavyHitters(HashMap<String, Pair<Long, Double>> heavyHitters, int num) {
StringBuilder sb = new StringBuilder();
@SuppressWarnings("unchecked")
Entry<String, Pair<Long, Double>>[] hhArr = heavyHitters.entrySet().toArray(new Entry[0]);
@@ -105,7 +222,7 @@ public class FederatedStatistics {
return e1.getValue().getRight().compareTo(e2.getValue().getRight());
}
});
-
+
sb.append("Heavy hitter instructions:\n");
final String numCol = "#";
final String instCol = "Instruction";
@@ -197,6 +314,33 @@ public class FederatedStatistics {
return retArr;
}
+
+ public static void incFedLookupTableGetCount() {
+ fedLookupTableGetCount.increment();
+ }
+
+ public static void incFedLookupTableGetTime(long time) {
+ fedLookupTableGetTime.add(time);
+ }
+
+ public static void incFedLookupTableEntryCount() {
+ fedLookupTableEntryCount.increment();
+ }
+
+ public static String displayFedLookupTableStats() {
+ if(fedLookupTableGetCount.longValue() > 0) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Fed LookupTable (Get, Entries):\t" +
+ fedLookupTableGetCount.longValue() + "/" +
+ fedLookupTableEntryCount.longValue() + ".\n");
+ // sb.append(String.format("Fed LookupTable Get Time:\t%.3f sec.\n",
+ // fedLookupTableGetTime.doubleValue() / 1000000000));
+ return sb.toString();
+ }
+ return "";
+ }
+
+
private static class FedStatsCollectFunction extends FederatedUDF {
private static final long serialVersionUID = 1L;
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/FederatedPSControlThread.java b/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/FederatedPSControlThread.java
index 30e90be..004e35b 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/FederatedPSControlThread.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/FederatedPSControlThread.java
@@ -56,7 +56,7 @@ import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.RightScalarOperator;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.util.ProgramConverter;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.ParamServStatistics;
import java.util.ArrayList;
import java.util.Collections;
@@ -442,8 +442,8 @@ public class FederatedPSControlThread extends PSWorker implements Callable<Void>
if(DMLScript.STATISTICS) {
long total = (long) tFedCommunication.stop();
long workerComputing = ((DoubleObject) responseData[1]).getLongValue();
- Statistics.accFedPSWorkerComputing(workerComputing);
- Statistics.accFedPSCommunicationTime(total - workerComputing);
+ ParamServStatistics.accFedWorkerComputing(workerComputing);
+ ParamServStatistics.accFedCommunicationTime(total - workerComputing);
}
return (ListObject) responseData[0];
}
@@ -580,7 +580,7 @@ public class FederatedPSControlThread extends PSWorker implements Callable<Void>
// Statistics methods
protected void accFedPSGradientWeightingTime(Timing time) {
if (DMLScript.STATISTICS && time != null)
- Statistics.accFedPSGradientWeightingTime((long) time.stop());
+ ParamServStatistics.accFedGradientWeightingTime((long) time.stop());
}
@Override
@@ -591,7 +591,7 @@ public class FederatedPSControlThread extends PSWorker implements Callable<Void>
@Override
protected void incWorkerNumber() {
if (DMLScript.STATISTICS)
- Statistics.incWorkerNumber();
+ ParamServStatistics.incWorkerNumber();
}
@Override
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/LocalPSWorker.java b/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/LocalPSWorker.java
index 93207f3..5343332 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/LocalPSWorker.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/LocalPSWorker.java
@@ -35,7 +35,7 @@ import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyze
import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
import org.apache.sysds.runtime.instructions.cp.ListObject;
import org.apache.sysds.runtime.util.CommonThreadPool;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.ParamServStatistics;
public class LocalPSWorker extends PSWorker implements Callable<Void> {
@@ -266,24 +266,24 @@ public class LocalPSWorker extends PSWorker implements Callable<Void> {
@Override
protected void incWorkerNumber() {
if (DMLScript.STATISTICS)
- Statistics.incWorkerNumber();
+ ParamServStatistics.incWorkerNumber();
}
@Override
protected void accLocalModelUpdateTime(Timing time) {
if (DMLScript.STATISTICS)
- Statistics.accPSLocalModelUpdateTime((long) time.stop());
+ ParamServStatistics.accLocalModelUpdateTime((long) time.stop());
}
@Override
protected void accBatchIndexingTime(Timing time) {
if (DMLScript.STATISTICS)
- Statistics.accPSBatchIndexingTime((long) time.stop());
+ ParamServStatistics.accBatchIndexingTime((long) time.stop());
}
@Override
protected void accGradientComputeTime(Timing time) {
if (DMLScript.STATISTICS)
- Statistics.accPSGradientComputeTime((long) time.stop());
+ ParamServStatistics.accGradientComputeTime((long) time.stop());
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/ParamServer.java b/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/ParamServer.java
index a27aafc..009dc20 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/ParamServer.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/ParamServer.java
@@ -49,7 +49,7 @@ import org.apache.sysds.runtime.instructions.cp.FunctionCallCPInstruction;
import org.apache.sysds.runtime.instructions.cp.ListObject;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.RightScalarOperator;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.ParamServStatistics;
public abstract class ParamServer
{
@@ -281,7 +281,7 @@ public abstract class ParamServer
Timing tAgg = DMLScript.STATISTICS ? new Timing(true) : null;
_model = updateLocalModel(_ec, gradients, _model);
if (DMLScript.STATISTICS && tAgg != null)
- Statistics.accPSAggregationTime((long) tAgg.stop());
+ ParamServStatistics.accAggregationTime((long) tAgg.stop());
}
/**
@@ -328,7 +328,7 @@ public abstract class ParamServer
if(allFinished()) {
_model = setParams(_ec, _accModels, _model);
if (DMLScript.STATISTICS && tAgg != null)
- Statistics.accPSAggregationTime((long) tAgg.stop());
+ ParamServStatistics.accAggregationTime((long) tAgg.stop());
_accModels = null; //reset for next accumulation
// This if has grown to be quite complex its function is rather simple. Validate at the end of each epoch
@@ -425,7 +425,7 @@ public abstract class ParamServer
//broadcast copy of model to specific worker, cleaned up by worker
_modelMap.get(workerID).put(ParamservUtils.copyList(_model, false));
if (DMLScript.STATISTICS && tBroad != null)
- Statistics.accPSModelBroadcastTime((long) tBroad.stop());
+ ParamServStatistics.accModelBroadcastTime((long) tBroad.stop());
}
/**
@@ -434,9 +434,9 @@ public abstract class ParamServer
private void time_epoch() {
if (DMLScript.STATISTICS) {
//TODO double check correctness with multiple, potentially concurrent paramserv invocation
- Statistics.accPSExecutionTime((long) Statistics.getPSExecutionTimer().stop());
- double current_total_execution_time = Statistics.getPSExecutionTime();
- double current_total_validation_time = Statistics.getPSValidationTime();
+ ParamServStatistics.accExecutionTime((long) ParamServStatistics.getExecutionTimer().stop());
+ double current_total_execution_time = ParamServStatistics.getExecutionTime();
+ double current_total_validation_time = ParamServStatistics.getValidationTime();
double time_to_epoch = current_total_execution_time - current_total_validation_time;
if (LOG.isInfoEnabled())
@@ -469,7 +469,7 @@ public abstract class ParamServer
LOG.info("[+] PARAMSERV: validation-loss: " + loss + " validation-accuracy: " + accuracy);
if(tValidate != null)
- Statistics.accPSValidationTime((long) tValidate.stop());
+ ParamServStatistics.accValidationTime((long) tValidate.stop());
}
public FunctionCallCPInstruction getAggInst() {
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/SparkParamservUtils.java b/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/SparkParamservUtils.java
index 3a40ad5..9dc075d 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/SparkParamservUtils.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/SparkParamservUtils.java
@@ -33,7 +33,7 @@ import org.apache.sysds.runtime.controlprogram.paramserv.dp.DataPartitionerSpark
import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.ParamServStatistics;
import scala.Tuple2;
@@ -112,7 +112,7 @@ public class SparkParamservUtils {
.mapToPair(new DataPartitionerSparkAggregator(features.getNumColumns(), labels.getNumColumns()));
if (DMLScript.STATISTICS)
- Statistics.accPSSetupTime((long) tSetup.stop());
+ ParamServStatistics.accSetupTime((long) tSetup.stop());
return result;
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptimizationWrapper.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptimizationWrapper.java
index 6188abb..0568198 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptimizationWrapper.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptimizationWrapper.java
@@ -54,7 +54,7 @@ import org.apache.sysds.runtime.controlprogram.parfor.stat.Stat;
import org.apache.sysds.runtime.controlprogram.parfor.stat.StatisticMonitor;
import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
import org.apache.sysds.runtime.util.UtilFunctions;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.ParForStatistics;
/**
@@ -124,7 +124,7 @@ public class OptimizationWrapper
//maintain statistics
if( DMLScript.STATISTICS )
- Statistics.incrementParForOptimCount();
+ ParForStatistics.incrementOptimCount();
//create specified optimizer
Optimizer opt = createOptimizer( otype );
@@ -242,7 +242,7 @@ public class OptimizationWrapper
long ltime = (long) time.stop();
LOG.trace("ParFOR Opt: Optimized plan in "+ltime+"ms.");
if( DMLScript.STATISTICS )
- Statistics.incrementParForOptimTime(ltime);
+ ParForStatistics.incrementOptimTime(ltime);
//monitor stats
if( monitor ) {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/ParamservBuiltinCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/ParamservBuiltinCPInstruction.java
index 1408c66..67fa582 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/ParamservBuiltinCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/ParamservBuiltinCPInstruction.java
@@ -88,7 +88,7 @@ import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyze
import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.util.ProgramConverter;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.ParamServStatistics;
public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruction {
private static final Log LOG = LogFactory.getLog(ParamservBuiltinCPInstruction.class.getName());
@@ -132,7 +132,7 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
private void runFederated(ExecutionContext ec) {
if(DMLScript.STATISTICS)
- Statistics.getPSExecutionTimer().start();
+ ParamServStatistics.getExecutionTimer().start();
Timing tSetup = DMLScript.STATISTICS ? new Timing(true) : null;
LOG.info("PARAMETER SERVER");
@@ -158,7 +158,7 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
LOG.info("[+] Seed: " + seed);
}
if (tSetup != null)
- Statistics.accPSSetupTime((long) tSetup.stop());
+ ParamServStatistics.accSetupTime((long) tSetup.stop());
// partition federated data
Timing tDataPartitioning = DMLScript.STATISTICS ? new Timing(true) : null;
@@ -166,7 +166,7 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
.doPartitioning(ec.getMatrixObject(getParam(PS_FEATURES)), ec.getMatrixObject(getParam(PS_LABELS)));
int workerNum = result._workerNum;
if (DMLScript.STATISTICS)
- Statistics.accFedPSDataPartitioningTime((long) tDataPartitioning.stop());
+ ParamServStatistics.accFedDataPartitioningTime((long) tDataPartitioning.stop());
if (DMLScript.STATISTICS)
@@ -206,7 +206,7 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
threads.get(i).setup(result._weightingFactors.get(i));
}
if (DMLScript.STATISTICS)
- Statistics.accPSSetupTime((long) tSetup.stop());
+ ParamServStatistics.accSetupTime((long) tSetup.stop());
try {
// Launch the worker threads and wait for completion
@@ -215,7 +215,7 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
// Fetch the final model from ps
ec.setVariable(output.getName(), ps.getResult());
if (DMLScript.STATISTICS)
- Statistics.accPSExecutionTime((long) Statistics.getPSExecutionTimer().stop());
+ ParamServStatistics.accExecutionTime((long) ParamServStatistics.getExecutionTimer().stop());
} catch (InterruptedException | ExecutionException e) {
throw new DMLRuntimeException("ParamservBuiltinCPInstruction: unknown error: ", e);
} finally {
@@ -275,7 +275,7 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
server.getPort(), aSetup, aWorker, aUpdate, aIndex, aGrad, aRPC, aBatch, aEpoch, nbatches, modelAvg);
if (DMLScript.STATISTICS)
- Statistics.accPSSetupTime((long) tSetup.stop());
+ ParamServStatistics.accSetupTime((long) tSetup.stop());
MatrixObject features = sec.getMatrixObject(getParam(PS_FEATURES));
MatrixObject labels = sec.getMatrixObject(getParam(PS_LABELS));
@@ -290,12 +290,12 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
// Accumulate the statistics for remote workers
if (DMLScript.STATISTICS) {
- Statistics.accPSSetupTime(aSetup.value());
- Statistics.incWorkerNumber(aWorker.value());
- Statistics.accPSLocalModelUpdateTime(aUpdate.value());
- Statistics.accPSBatchIndexingTime(aIndex.value());
- Statistics.accPSGradientComputeTime(aGrad.value());
- Statistics.accPSRpcRequestTime(aRPC.value());
+ ParamServStatistics.accSetupTime(aSetup.value());
+ ParamServStatistics.incWorkerNumber(aWorker.value());
+ ParamServStatistics.accLocalModelUpdateTime(aUpdate.value());
+ ParamServStatistics.accBatchIndexingTime(aIndex.value());
+ ParamServStatistics.accGradientComputeTime(aGrad.value());
+ ParamServStatistics.accRpcRequestTime(aRPC.value());
}
// Fetch the final model from ps
@@ -304,7 +304,7 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
private void runLocally(ExecutionContext ec, PSModeType mode) {
if(DMLScript.STATISTICS)
- Statistics.getPSExecutionTimer().start();
+ ParamServStatistics.getExecutionTimer().start();
Timing tSetup = DMLScript.STATISTICS ? new Timing(true) : null;
int workerNum = getWorkerNum(mode);
@@ -350,7 +350,7 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
partitionLocally(scheme, ec, workers);
if (DMLScript.STATISTICS)
- Statistics.accPSSetupTime((long) tSetup.stop());
+ ParamServStatistics.accSetupTime((long) tSetup.stop());
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("\nConfiguration of paramserv func: "
@@ -366,7 +366,7 @@ public class ParamservBuiltinCPInstruction extends ParameterizedBuiltinCPInstruc
// Fetch the final model from ps
ec.setVariable(output.getName(), ps.getResult());
if (DMLScript.STATISTICS)
- Statistics.accPSExecutionTime((long) Statistics.getPSExecutionTimer().stop());
+ ParamServStatistics.accExecutionTime((long) ParamServStatistics.getExecutionTimer().stop());
} catch (InterruptedException | ExecutionException e) {
throw new DMLRuntimeException("ParamservBuiltinCPInstruction: some error occurred: ", e);
} finally {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/TriggerBroadcastTask.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/TriggerBroadcastTask.java
index 122648e..c2f01c3 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/TriggerBroadcastTask.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/TriggerBroadcastTask.java
@@ -23,7 +23,7 @@ import org.apache.sysds.api.DMLScript;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.controlprogram.context.SparkExecutionContext;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.SparkStatistics;
public class TriggerBroadcastTask implements Runnable {
ExecutionContext _ec;
@@ -46,7 +46,7 @@ public class TriggerBroadcastTask implements Runnable {
//TODO: Count only if successful (owned lock)
if (DMLScript.STATISTICS)
- Statistics.incSparkAsyncBroadcastCount(1);
+ SparkStatistics.incAsyncBroadcastCount(1);
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/TriggerRemoteOperationsTask.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/TriggerRemoteOperationsTask.java
index 6eea8c9..255a2d6 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/TriggerRemoteOperationsTask.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/TriggerRemoteOperationsTask.java
@@ -21,7 +21,8 @@ package org.apache.sysds.runtime.instructions.cp;
import org.apache.sysds.api.DMLScript;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.runtime.controlprogram.federated.FederatedStatistics;
+import org.apache.sysds.utils.stats.SparkStatistics;
public class TriggerRemoteOperationsTask implements Runnable {
MatrixObject _prefetchMO;
@@ -45,9 +46,9 @@ public class TriggerRemoteOperationsTask implements Runnable {
}
if (DMLScript.STATISTICS && prefetched) {
if (_prefetchMO.isFederated())
- Statistics.incFedAsyncPrefetchCount(1);
+ FederatedStatistics.incAsyncPrefetchCount(1);
else
- Statistics.incSparkAsyncPrefetchCount(1);
+ SparkStatistics.incAsyncPrefetchCount(1);
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/InitFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/InitFEDInstruction.java
index b4d2e04..3db1c8b 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/InitFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/InitFEDInstruction.java
@@ -31,6 +31,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.apache.sysds.api.DMLScript;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.logging.Log;
@@ -122,8 +123,9 @@ public class InitFEDInstruction extends FEDInstruction implements LineageTraceab
int port = Integer.parseInt(parsedValues[1]);
String filePath = parsedValues[2];
- // register the federated worker for federated statistics creation
- FederatedStatistics.registerFedWorker(host, port);
+ if(DMLScript.FED_STATISTICS)
+ // register the federated worker for federated statistics creation
+ FederatedStatistics.registerFedWorker(host, port);
// get beginning and end of data ranges
List<Data> rangesData = ranges.getData();
diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixDNNConv2d.java b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixDNNConv2d.java
index 575a109..0c75434 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixDNNConv2d.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixDNNConv2d.java
@@ -27,7 +27,7 @@ import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.matrix.data.LibMatrixDNNRotate180.Rotate180Worker;
import org.apache.sysds.utils.NativeHelper;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.NativeStatistics;
/**
* This class contains the set of operators used for performing conv2d
@@ -55,8 +55,8 @@ public class LibMatrixDNNConv2d
boolean applyNative = isEligibleForConv2dSparse(params)
&& !(!isEmptyDenseInput && isTransPref);
if( applyNative )
- Statistics.numNativeSparseConv2dCalls.increment();
-
+ NativeStatistics.incrementNumSparseConv2dCalls();
+
//transpose filter once for efficient sparse-dense multiplies in LoopedIm2ColConv2dTransAllChan
//in order to share the temporary object and its creation costs across threads
if( !applyNative && !isEmptyDenseInput && isTransPref ) {
@@ -98,7 +98,7 @@ public class LibMatrixDNNConv2d
boolean applyNative = isEligibleForConv2dBackwardFilterSparseDense(params)
&& !params.input2.isInSparseFormat();
if( applyNative )
- Statistics.numNativeSparseConv2dBwdFilterCalls.increment();
+ NativeStatistics.incrementNumSparseConv2dBwdFilterCalls();
for(int i = 0; i*taskSize < params.N; i++) {
//note: we prefer the java backend for sparse filters because the native
@@ -135,7 +135,7 @@ public class LibMatrixDNNConv2d
boolean applyNative = isEligibleForConv2dBackwardDataDense(params)
&& !params.input2.isInSparseFormat();
if( applyNative )
- Statistics.numNativeSparseConv2dBwdDataCalls.increment();
+ NativeStatistics.incrementNumSparseConv2dBwdDataCalls();
for(int i = 0; i*taskSize < params.N; i++) {
//note: we prefer the java backend for sparse filters because the native
diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixNative.java b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixNative.java
index ac0b069..4c8dc98 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixNative.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixNative.java
@@ -33,7 +33,7 @@ import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.runtime.data.DenseBlock;
import org.apache.sysds.runtime.data.DenseBlockFactory;
import org.apache.sysds.utils.NativeHelper;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.NativeStatistics;
public class LibMatrixNative
{
@@ -123,15 +123,15 @@ public class LibMatrixNative
if(nnz > -1) {
if(DMLScript.STATISTICS) {
- Statistics.nativeLibMatrixMultTime += System.nanoTime() - start;
- Statistics.numNativeLibMatrixMultCalls.increment();
+ NativeStatistics.incrementLibMatrixMultTime(System.nanoTime() - start);
+ NativeStatistics.incrementNumLibMatrixMultCalls();
}
ret.setNonZeros(nnz);
ret.examSparsity();
return ret;
}
//else record failure and fallback to java
- Statistics.incrementNativeFailuresCounter();
+ NativeStatistics.incrementFailuresCounter();
LOG.warn("matrixMult: Native mat mult failed. Falling back to java version ("
+ "loaded=" + NativeHelper.isNativeLibraryLoaded()
+ ", sparse=" + (m1.isInSparseFormat() | m2.isInSparseFormat()) + ")");
@@ -178,8 +178,8 @@ public class LibMatrixNative
if(nnz > -1) {
if(DMLScript.STATISTICS) {
- Statistics.nativeLibMatrixMultTime += System.nanoTime() - start;
- Statistics.numNativeLibMatrixMultCalls.increment();
+ NativeStatistics.incrementLibMatrixMultTime(System.nanoTime() - start);
+ NativeStatistics.incrementNumLibMatrixMultCalls();
}
ret.setNonZeros(nnz);
ret.examSparsity();
@@ -187,7 +187,7 @@ public class LibMatrixNative
}
//fallback to default java implementation
LOG.warn("Native TSMM failed. Falling back to java version.");
- Statistics.incrementNativeFailuresCounter();
+ NativeStatistics.incrementFailuresCounter();
}
if( k > 1 )
LibMatrixMult.matrixMultTransposeSelf(m1, ret, leftTrans, k);
@@ -246,8 +246,8 @@ public class LibMatrixNative
//post processing and error handling
if(nnz != -1) {
if(DMLScript.STATISTICS) {
- Statistics.nativeConv2dTime += System.nanoTime() - start;
- Statistics.numNativeConv2dCalls.increment();
+ NativeStatistics.incrementConv2dTime(System.nanoTime() - start);
+ NativeStatistics.incrementNumConv2dCalls();
}
outputBlock.setNonZeros(nnz);
return;
@@ -257,7 +257,7 @@ public class LibMatrixNative
LOG.warn("Native conv2d call returned with error - falling back to java operator.");
if( !(isSinglePrecision() && params.bias!=null) )
outputBlock.reset();
- Statistics.incrementNativeFailuresCounter();
+ NativeStatistics.incrementFailuresCounter();
}
}
@@ -291,8 +291,8 @@ public class LibMatrixNative
params.P, params.Q, params.numThreads);
if(nnz != -1) {
if(DMLScript.STATISTICS) {
- Statistics.nativeConv2dBwdFilterTime += System.nanoTime() - start;
- Statistics.numNativeConv2dBwdFilterCalls.increment();
+ NativeStatistics.incrementConv2dBwdFilterTime(System.nanoTime() - start);
+ NativeStatistics.incrementNumConv2dBwdFilterCalls();
}
// post-processing: maintain nnz
outputBlock.setNonZeros(nnz);
@@ -300,7 +300,7 @@ public class LibMatrixNative
}
else {
// Fall back to Java when failures
- Statistics.incrementNativeFailuresCounter();
+ NativeStatistics.incrementFailuresCounter();
}
}
// Fall back to Java when failures or sparse
@@ -327,8 +327,8 @@ public class LibMatrixNative
params.P, params.Q, params.numThreads);
if(nnz != -1) {
if(DMLScript.STATISTICS) {
- Statistics.nativeConv2dBwdDataTime += System.nanoTime() - start;
- Statistics.numNativeConv2dBwdDataCalls.increment();
+ NativeStatistics.incrementConv2dBwdDataTime(System.nanoTime() - start);
+ NativeStatistics.incrementNumConv2dBwdDataCalls();
}
// post-processing: maintain nnz
outputBlock.setNonZeros(nnz);
@@ -336,7 +336,7 @@ public class LibMatrixNative
}
else {
// Fall back to Java when failures
- Statistics.incrementNativeFailuresCounter();
+ NativeStatistics.incrementFailuresCounter();
}
}
// Fall back to Java when failures or sparse
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoder.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoder.java
index 6c9ac6a..f98d323 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoder.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoder.java
@@ -47,7 +47,7 @@ import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.DependencyTask;
import org.apache.sysds.runtime.util.DependencyThreadPool;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
/**
* Base class for all transform encoders providing both a row and block interface for decoding frames to matrices.
@@ -95,19 +95,19 @@ public abstract class ColumnEncoder implements Encoder, Comparable<ColumnEncoder
long t = System.nanoTime()-t0;
switch (this.getTransformType()){
case RECODE:
- Statistics.incTransformRecodeApplyTime(t);
+ TransformStatistics.incRecodeApplyTime(t);
break;
case BIN:
- Statistics.incTransformBinningApplyTime(t);
+ TransformStatistics.incBinningApplyTime(t);
break;
case DUMMYCODE:
- Statistics.incTransformDummyCodeApplyTime(t);
+ TransformStatistics.incDummyCodeApplyTime(t);
break;
case FEATURE_HASH:
- Statistics.incTransformFeatureHashingApplyTime(t);
+ TransformStatistics.incFeatureHashingApplyTime(t);
break;
case PASS_THROUGH:
- Statistics.incTransformPassThroughApplyTime(t);
+ TransformStatistics.incPassThroughApplyTime(t);
break;
default:
break;
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderBin.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderBin.java
index d802254..005c2c2 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderBin.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderBin.java
@@ -34,7 +34,7 @@ import org.apache.sysds.lops.Lop;
import org.apache.sysds.runtime.controlprogram.caching.CacheBlock;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
public class ColumnEncoderBin extends ColumnEncoder {
public static final String MIN_PREFIX = "min";
@@ -91,7 +91,7 @@ public class ColumnEncoderBin extends ColumnEncoder {
double[] pairMinMax = getMinMaxOfCol(in, _colID, 0, -1);
computeBins(pairMinMax[0], pairMinMax[1]);
if(DMLScript.STATISTICS)
- Statistics.incTransformBinningBuildTime(System.nanoTime()-t0);
+ TransformStatistics.incBinningBuildTime(System.nanoTime()-t0);
}
protected double getCode(CacheBlock in, int row){
@@ -307,7 +307,7 @@ public class ColumnEncoderBin extends ColumnEncoder {
return null;
_encoder.applySparse(_input, _out, _outputCol, _startRow, _blk);
if(DMLScript.STATISTICS)
- Statistics.incTransformBinningApplyTime(System.nanoTime()-t0);
+ TransformStatistics.incBinningApplyTime(System.nanoTime()-t0);
return null;
}
@@ -343,7 +343,7 @@ public class ColumnEncoderBin extends ColumnEncoder {
_partialMinMax.put(_startRow, minMax);
}
if (DMLScript.STATISTICS)
- Statistics.incTransformBinningBuildTime(System.nanoTime()-t0);
+ TransformStatistics.incBinningBuildTime(System.nanoTime()-t0);
return null;
}
@@ -374,7 +374,7 @@ public class ColumnEncoderBin extends ColumnEncoder {
_encoder.computeBins(min, max);
if(DMLScript.STATISTICS)
- Statistics.incTransformBinningBuildTime(System.nanoTime()-t0);
+ TransformStatistics.incBinningBuildTime(System.nanoTime()-t0);
return null;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderDummycode.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderDummycode.java
index 5c1cd11..01e7d5c 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderDummycode.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderDummycode.java
@@ -37,7 +37,7 @@ import org.apache.sysds.runtime.data.SparseBlockCSR;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.DependencyTask;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
public class ColumnEncoderDummycode extends ColumnEncoder {
private static final long serialVersionUID = 5832130477659116489L;
@@ -292,7 +292,7 @@ public class ColumnEncoderDummycode extends ColumnEncoder {
return null;
_encoder.applySparse(_input, _out, _outputCol, _startRow, _blk);
if (DMLScript.STATISTICS)
- Statistics.incTransformDummyCodeApplyTime(System.nanoTime()-t0);
+ TransformStatistics.incDummyCodeApplyTime(System.nanoTime()-t0);
return null;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderFeatureHash.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderFeatureHash.java
index f7d9033..9445474 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderFeatureHash.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderFeatureHash.java
@@ -31,7 +31,7 @@ import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.DependencyTask;
import org.apache.sysds.runtime.util.UtilFunctions;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
/**
* Class used for feature hashing transformation of frames.
@@ -168,7 +168,7 @@ public class ColumnEncoderFeatureHash extends ColumnEncoder {
long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
_encoder.applySparse(_input, _out, _outputCol, _startRow, _blk);
if(DMLScript.STATISTICS)
- Statistics.incTransformFeatureHashingApplyTime(System.nanoTime()-t0);
+ TransformStatistics.incFeatureHashingApplyTime(System.nanoTime()-t0);
return null;
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderPassThrough.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderPassThrough.java
index 36784ab..5ceba09 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderPassThrough.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderPassThrough.java
@@ -33,7 +33,7 @@ import org.apache.sysds.runtime.data.SparseRowVector;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.DependencyTask;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
public class ColumnEncoderPassThrough extends ColumnEncoder {
private static final long serialVersionUID = -8473768154646831882L;
@@ -170,7 +170,7 @@ public class ColumnEncoderPassThrough extends ColumnEncoder {
long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
_encoder.applySparse(_input, _out, _outputCol, _startRow, _blk);
if(DMLScript.STATISTICS)
- Statistics.incTransformPassThroughApplyTime(System.nanoTime()-t0);
+ TransformStatistics.incPassThroughApplyTime(System.nanoTime()-t0);
return null;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderRecode.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderRecode.java
index 14c0b61..3afce5a 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderRecode.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderRecode.java
@@ -37,7 +37,7 @@ import org.apache.sysds.lops.Lop;
import org.apache.sysds.runtime.controlprogram.caching.CacheBlock;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
public class ColumnEncoderRecode extends ColumnEncoder {
private static final long serialVersionUID = 8213163881283341874L;
@@ -140,7 +140,7 @@ public class ColumnEncoderRecode extends ColumnEncoder {
long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
makeRcdMap(in, _rcdMap, _colID, 0, in.getNumRows());
if(DMLScript.STATISTICS){
- Statistics.incTransformRecodeBuildTime(System.nanoTime() - t0);
+ TransformStatistics.incRecodeBuildTime(System.nanoTime() - t0);
}
}
@@ -347,7 +347,7 @@ public class ColumnEncoderRecode extends ColumnEncoder {
return null;
_encoder.applySparse(_input, _out, _outputCol, _startRow, _blk);
if(DMLScript.STATISTICS){
- Statistics.incTransformRecodeApplyTime(System.nanoTime() - t0);
+ TransformStatistics.incRecodeApplyTime(System.nanoTime() - t0);
}
return null;
}
@@ -388,7 +388,7 @@ public class ColumnEncoderRecode extends ColumnEncoder {
_partialMaps.put(_startRow, partialMap);
}
if(DMLScript.STATISTICS){
- Statistics.incTransformRecodeBuildTime(System.nanoTime() - t0);
+ TransformStatistics.incRecodeBuildTime(System.nanoTime() - t0);
}
return null;
}
@@ -421,7 +421,7 @@ public class ColumnEncoderRecode extends ColumnEncoder {
});
_encoder._rcdMap = rcdMap;
if(DMLScript.STATISTICS){
- Statistics.incTransformRecodeBuildTime(System.nanoTime() - t0);
+ TransformStatistics.incRecodeBuildTime(System.nanoTime() - t0);
}
return null;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderFactory.java b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderFactory.java
index 012379a..67c0ea1 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderFactory.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderFactory.java
@@ -28,7 +28,7 @@ import org.apache.sysds.runtime.transform.TfUtils.TfMethod;
import org.apache.sysds.runtime.transform.encode.ColumnEncoder.EncoderType;
import org.apache.sysds.runtime.transform.meta.TfMetaUtils;
import org.apache.sysds.runtime.util.UtilFunctions;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
import org.apache.wink.json4j.JSONArray;
import org.apache.wink.json4j.JSONObject;
@@ -128,21 +128,21 @@ public class EncoderFactory {
// create composite decoder of all created encoders
for(Entry<Integer, List<ColumnEncoder>> listEntry : colEncoders.entrySet()) {
if(DMLScript.STATISTICS)
- Statistics.incTransformEncoderCount(listEntry.getValue().size());
+ TransformStatistics.incEncoderCount(listEntry.getValue().size());
lencoders.add(new ColumnEncoderComposite(listEntry.getValue()));
}
encoder = new MultiColumnEncoder(lencoders);
if(!oIDs.isEmpty()) {
encoder.addReplaceLegacyEncoder(new EncoderOmit(jSpec, colnames, schema.length, minCol, maxCol));
if(DMLScript.STATISTICS)
- Statistics.incTransformEncoderCount(1);
+ TransformStatistics.incEncoderCount(1);
}
if(!mvIDs.isEmpty()) {
EncoderMVImpute ma = new EncoderMVImpute(jSpec, colnames, schema.length, minCol, maxCol);
ma.initRecodeIDList(rcIDs);
encoder.addReplaceLegacyEncoder(ma);
if(DMLScript.STATISTICS)
- Statistics.incTransformEncoderCount(1);
+ TransformStatistics.incEncoderCount(1);
}
// initialize meta data w/ robustness for superset of cols
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderMVImpute.java b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderMVImpute.java
index f77e690..63950fa 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderMVImpute.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderMVImpute.java
@@ -31,7 +31,7 @@ import org.apache.sysds.runtime.transform.TfUtils.TfMethod;
import org.apache.sysds.runtime.transform.meta.TfMetaUtils;
import org.apache.sysds.runtime.util.IndexRange;
import org.apache.sysds.runtime.util.UtilFunctions;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
import org.apache.wink.json4j.JSONArray;
import org.apache.wink.json4j.JSONException;
import org.apache.wink.json4j.JSONObject;
@@ -219,7 +219,7 @@ public class EncoderMVImpute extends LegacyEncoder {
throw new RuntimeException(ex);
}
if(DMLScript.STATISTICS)
- Statistics.incTransformImputeBuildTime(System.nanoTime()-t0);
+ TransformStatistics.incImputeBuildTime(System.nanoTime()-t0);
}
@Override
@@ -233,7 +233,7 @@ public class EncoderMVImpute extends LegacyEncoder {
}
}
if(DMLScript.STATISTICS)
- Statistics.incTransformImputeApplyTime(System.nanoTime()-t0);
+ TransformStatistics.incImputeApplyTime(System.nanoTime()-t0);
return out;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderOmit.java b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderOmit.java
index c2f3b68..68627dc 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderOmit.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderOmit.java
@@ -30,7 +30,7 @@ import org.apache.sysds.runtime.transform.TfUtils.TfMethod;
import org.apache.sysds.runtime.transform.meta.TfMetaUtils;
import org.apache.sysds.runtime.util.IndexRange;
import org.apache.sysds.runtime.util.UtilFunctions;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
import org.apache.wink.json4j.JSONException;
import org.apache.wink.json4j.JSONObject;
@@ -152,7 +152,7 @@ public class EncoderOmit extends LegacyEncoder {
_rmRows = rmRows;
if(DMLScript.STATISTICS)
- Statistics.incTransformOmitApplyTime(System.nanoTime()-t0);
+ TransformStatistics.incOmitApplyTime(System.nanoTime()-t0);
return ret;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/MultiColumnEncoder.java b/src/main/java/org/apache/sysds/runtime/transform/encode/MultiColumnEncoder.java
index 3d1d98e..09ecf64 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/MultiColumnEncoder.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/MultiColumnEncoder.java
@@ -58,7 +58,7 @@ import org.apache.sysds.runtime.util.DependencyTask;
import org.apache.sysds.runtime.util.DependencyThreadPool;
import org.apache.sysds.runtime.util.DependencyWrapperTask;
import org.apache.sysds.runtime.util.IndexRange;
-import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
public class MultiColumnEncoder implements Encoder {
@@ -400,7 +400,7 @@ public class MultiColumnEncoder implements Encoder {
if(DMLScript.STATISTICS) {
LOG.debug("Elapsed time for allocation: "+ ((double) System.nanoTime() - t0) / 1000000 + " ms");
- Statistics.incTransformOutMatrixPreProcessingTime(System.nanoTime()-t0);
+ TransformStatistics.incOutMatrixPreProcessingTime(System.nanoTime()-t0);
}
}
@@ -420,7 +420,7 @@ public class MultiColumnEncoder implements Encoder {
}
output.recomputeNonZeros();
if(DMLScript.STATISTICS)
- Statistics.incTransformOutMatrixPostProcessingTime(System.nanoTime()-t0);
+ TransformStatistics.incOutMatrixPostProcessingTime(System.nanoTime()-t0);
}
@Override
diff --git a/src/main/java/org/apache/sysds/utils/Statistics.java b/src/main/java/org/apache/sysds/utils/Statistics.java
index a3da2a7..12748a0 100644
--- a/src/main/java/org/apache/sysds/utils/Statistics.java
+++ b/src/main/java/org/apache/sysds/utils/Statistics.java
@@ -25,10 +25,7 @@ import org.apache.sysds.api.DMLScript;
import org.apache.sysds.conf.ConfigurationManager;
import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.runtime.controlprogram.caching.CacheStatistics;
-import org.apache.sysds.runtime.controlprogram.context.SparkExecutionContext;
-import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest.RequestType;
import org.apache.sysds.runtime.controlprogram.federated.FederatedStatistics;
-import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
import org.apache.sysds.runtime.instructions.Instruction;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.FunctionCallCPInstruction;
@@ -36,6 +33,13 @@ import org.apache.sysds.runtime.instructions.spark.SPInstruction;
import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
import org.apache.sysds.runtime.lineage.LineageCacheStatistics;
import org.apache.sysds.runtime.privacy.CheckedConstraintsLog;
+import org.apache.sysds.utils.stats.CodegenStatistics;
+import org.apache.sysds.utils.stats.RecompileStatistics;
+import org.apache.sysds.utils.stats.NativeStatistics;
+import org.apache.sysds.utils.stats.ParamServStatistics;
+import org.apache.sysds.utils.stats.ParForStatistics;
+import org.apache.sysds.utils.stats.SparkStatistics;
+import org.apache.sysds.utils.stats.TransformStatistics;
import java.lang.management.CompilationMXBean;
import java.lang.management.GarbageCollectorMXBean;
@@ -86,125 +90,20 @@ public class Statistics
private static long jitCompileTime = 0; //in milli sec
private static long jvmGCTime = 0; //in milli sec
private static long jvmGCCount = 0; //count
-
- //HOP DAG recompile stats (potentially high update frequency)
- private static final LongAdder hopRecompileTime = new LongAdder(); //in nano sec
- private static final LongAdder hopRecompilePred = new LongAdder(); //count
- private static final LongAdder hopRecompileSB = new LongAdder(); //count
-
- //CODEGEN
- private static final LongAdder codegenCompileTime = new LongAdder(); //in nano
- private static final LongAdder codegenClassCompileTime = new LongAdder(); //in nano
- private static final LongAdder codegenHopCompile = new LongAdder(); //count
- private static final LongAdder codegenCPlanCompile = new LongAdder(); //count
- private static final LongAdder codegenClassCompile = new LongAdder(); //count
- private static final LongAdder codegenEnumAll = new LongAdder(); //count
- private static final LongAdder codegenEnumAllP = new LongAdder(); //count
- private static final LongAdder codegenEnumEval = new LongAdder(); //count
- private static final LongAdder codegenEnumEvalP = new LongAdder(); //count
- private static final LongAdder codegenOpCacheHits = new LongAdder(); //count
- private static final LongAdder codegenOpCacheTotal = new LongAdder(); //count
- private static final LongAdder codegenPlanCacheHits = new LongAdder(); //count
- private static final LongAdder codegenPlanCacheTotal = new LongAdder(); //count
-
+
//Function recompile stats
private static final LongAdder funRecompileTime = new LongAdder(); //in nano sec
private static final LongAdder funRecompiles = new LongAdder(); //count
-
- //Spark-specific stats
- private static long sparkCtxCreateTime = 0;
- private static final LongAdder sparkParallelize = new LongAdder();
- private static final LongAdder sparkParallelizeCount = new LongAdder();
- private static final LongAdder sparkCollect = new LongAdder();
- private static final LongAdder sparkCollectCount = new LongAdder();
- private static final LongAdder sparkBroadcast = new LongAdder();
- private static final LongAdder sparkBroadcastCount = new LongAdder();
- private static final LongAdder sparkAsyncPrefetchCount = new LongAdder();
- private static final LongAdder sparkAsyncBroadcastCount = new LongAdder();
-
- // Paramserv function stats (time is in milli sec)
- private static final Timing psExecutionTimer = new Timing(false);
- private static final LongAdder psExecutionTime = new LongAdder();
- private static final LongAdder psNumWorkers = new LongAdder();
- private static final LongAdder psSetupTime = new LongAdder();
- private static final LongAdder psGradientComputeTime = new LongAdder();
- private static final LongAdder psAggregationTime = new LongAdder();
- private static final LongAdder psLocalModelUpdateTime = new LongAdder();
- private static final LongAdder psModelBroadcastTime = new LongAdder();
- private static final LongAdder psBatchIndexTime = new LongAdder();
- private static final LongAdder psRpcRequestTime = new LongAdder();
- private static final LongAdder psValidationTime = new LongAdder();
- // Federated parameter server specifics (time is in milli sec)
- private static final LongAdder fedPSDataPartitioningTime = new LongAdder();
- private static final LongAdder fedPSWorkerComputingTime = new LongAdder();
- private static final LongAdder fedPSGradientWeightingTime = new LongAdder();
- private static final LongAdder fedPSCommunicationTime = new LongAdder();
-
- //PARFOR optimization stats (low frequency updates)
- private static long parforOptTime = 0; //in milli sec
- private static long parforOptCount = 0; //count
- private static long parforInitTime = 0; //in milli sec
- private static long parforMergeTime = 0; //in milli sec
private static final LongAdder lTotalUIPVar = new LongAdder();
private static final LongAdder lTotalLix = new LongAdder();
private static final LongAdder lTotalLixUIP = new LongAdder();
- // Transformencode stats
- private static final LongAdder transformEncoderCount = new LongAdder();
-
- //private static final LongAdder transformBuildTime = new LongAdder();
- private static final LongAdder transformRecodeBuildTime = new LongAdder();
- private static final LongAdder transformBinningBuildTime = new LongAdder();
- private static final LongAdder transformImputeBuildTime = new LongAdder();
-
- //private static final LongAdder transformApplyTime = new LongAdder();
- private static final LongAdder transformRecodeApplyTime = new LongAdder();
- private static final LongAdder transformDummyCodeApplyTime = new LongAdder();
- private static final LongAdder transformPassThroughApplyTime = new LongAdder();
- private static final LongAdder transformFeatureHashingApplyTime = new LongAdder();
- private static final LongAdder transformBinningApplyTime = new LongAdder();
- private static final LongAdder transformOmitApplyTime = new LongAdder();
- private static final LongAdder transformImputeApplyTime = new LongAdder();
-
-
- private static final LongAdder transformOutMatrixPreProcessingTime = new LongAdder();
- private static final LongAdder transformOutMatrixPostProcessingTime = new LongAdder();
-
- // Federated stats
- private static final LongAdder federatedReadCount = new LongAdder();
- private static final LongAdder federatedPutCount = new LongAdder();
- private static final LongAdder federatedGetCount = new LongAdder();
- private static final LongAdder federatedExecuteInstructionCount = new LongAdder();
- private static final LongAdder federatedExecuteUDFCount = new LongAdder();
- private static final LongAdder fedAsyncPrefetchCount = new LongAdder();
-
- private static LongAdder numNativeFailures = new LongAdder();
- public static LongAdder numNativeLibMatrixMultCalls = new LongAdder();
- public static LongAdder numNativeConv2dCalls = new LongAdder();
- public static LongAdder numNativeConv2dBwdDataCalls = new LongAdder();
- public static LongAdder numNativeConv2dBwdFilterCalls = new LongAdder();
- public static LongAdder numNativeSparseConv2dCalls = new LongAdder();
- public static LongAdder numNativeSparseConv2dBwdFilterCalls = new LongAdder();
- public static LongAdder numNativeSparseConv2dBwdDataCalls = new LongAdder();
- public static long nativeLibMatrixMultTime = 0;
- public static long nativeConv2dTime = 0;
- public static long nativeConv2dBwdDataTime = 0;
- public static long nativeConv2dBwdFilterTime = 0;
-
public static long recomputeNNZTime = 0;
public static long examSparsityTime = 0;
public static long allocateDoubleArrTime = 0;
public static boolean allowWorkerStatistics = true;
-
- public static void incrementNativeFailuresCounter() {
- numNativeFailures.increment();
- // This is very rare and am not sure it is possible at all. Our initial experiments never encountered this case.
- // Note: all the native calls have a fallback to Java; so if the user wants she can recompile SystemDS by
- // commenting this exception and everything should work fine.
- throw new RuntimeException("Unexpected ERROR: OOM caused during JNI transfer. Please disable native BLAS by setting enviroment variable: SYSTEMDS_BLAS=none");
- }
public static long getNoOfExecutedSPInst() {
return numExecutedSPInst.longValue();
@@ -225,11 +124,7 @@ public class Statistics
public static void incrementNoOfCompiledSPInst() {
numCompiledSPInst.increment();
}
-
- public static boolean createdSparkContext() {
- return sparkCtxCreateTime > 0;
- }
-
+
public static long getTotalUIPVar() {
return lTotalUIPVar.longValue();
}
@@ -279,124 +174,6 @@ public class Statistics
public static synchronized void incrementJVMgcCount( long delta ) {
jvmGCCount += delta;
}
-
- public static void incrementHOPRecompileTime( long delta ) {
- hopRecompileTime.add(delta);
- }
-
- public static void incrementHOPRecompilePred() {
- hopRecompilePred.increment();
- }
-
- public static void incrementHOPRecompilePred(long delta) {
- hopRecompilePred.add(delta);
- }
-
- public static void incrementHOPRecompileSB() {
- hopRecompileSB.increment();
- }
-
- public static void incrementHOPRecompileSB(long delta) {
- hopRecompileSB.add(delta);
- }
-
- public static void incrementCodegenDAGCompile() {
- codegenHopCompile.increment();
- }
-
- public static void incrementCodegenCPlanCompile(long delta) {
- codegenCPlanCompile.add(delta);
- }
-
- public static void incrementCodegenEnumAll(long delta) {
- codegenEnumAll.add(delta);
- }
- public static void incrementCodegenEnumAllP(long delta) {
- codegenEnumAllP.add(delta);
- }
- public static void incrementCodegenEnumEval(long delta) {
- codegenEnumEval.add(delta);
- }
- public static void incrementCodegenEnumEvalP(long delta) {
- codegenEnumEvalP.add(delta);
- }
-
- public static void incrementCodegenClassCompile() {
- codegenClassCompile.increment();
- }
-
- public static void incrementCodegenCompileTime(long delta) {
- codegenCompileTime.add(delta);
- }
-
- public static void incrementCodegenClassCompileTime(long delta) {
- codegenClassCompileTime.add(delta);
- }
-
- public static void incrementCodegenOpCacheHits() {
- codegenOpCacheHits.increment();
- }
-
- public static void incrementCodegenOpCacheTotal() {
- codegenOpCacheTotal.increment();
- }
-
- public static void incrementCodegenPlanCacheHits() {
- codegenPlanCacheHits.increment();
- }
-
- public static void incrementCodegenPlanCacheTotal() {
- codegenPlanCacheTotal.increment();
- }
-
- public static long getCodegenDAGCompile() {
- return codegenHopCompile.longValue();
- }
-
- public static long getCodegenCPlanCompile() {
- return codegenCPlanCompile.longValue();
- }
-
- public static long getCodegenEnumAll() {
- return codegenEnumAll.longValue();
- }
- public static long getCodegenEnumAllP() {
- return codegenEnumAllP.longValue();
- }
- public static long getCodegenEnumEval() {
- return codegenEnumEval.longValue();
- }
- public static long getCodegenEnumEvalP() {
- return codegenEnumEvalP.longValue();
- }
-
- public static long getCodegenClassCompile() {
- return codegenClassCompile.longValue();
- }
-
- public static long getCodegenCompileTime() {
- return codegenCompileTime.longValue();
- }
-
- public static long getCodegenClassCompileTime() {
- return codegenClassCompileTime.longValue();
- }
-
- public static long getCodegenOpCacheHits() {
- return codegenOpCacheHits.longValue();
- }
-
- public static long getCodegenOpCacheTotal() {
- return codegenOpCacheTotal.longValue();
- }
-
- public static long getCodegenPlanCacheHits() {
- return codegenPlanCacheHits.longValue();
- }
-
- public static long getCodegenPlanCacheTotal() {
- return codegenPlanCacheTotal.longValue();
- }
public static void incrementFunRecompileTime( long delta ) {
funRecompileTime.add(delta);
@@ -405,48 +182,6 @@ public class Statistics
public static void incrementFunRecompiles() {
funRecompiles.increment();
}
-
- public static synchronized void incrementParForOptimCount(){
- parforOptCount ++;
- }
-
- public static synchronized void incrementParForOptimTime( long time ) {
- parforOptTime += time;
- }
-
- public static synchronized void incrementParForInitTime( long time ) {
- parforInitTime += time;
- }
-
- public static synchronized void incrementParForMergeTime( long time ) {
- parforMergeTime += time;
- }
-
- public static synchronized void incFederated(RequestType rqt){
- switch (rqt) {
- case READ_VAR:
- federatedReadCount.increment();
- break;
- case PUT_VAR:
- federatedPutCount.increment();
- break;
- case GET_VAR:
- federatedGetCount.increment();
- break;
- case EXEC_INST:
- federatedExecuteInstructionCount.increment();
- break;
- case EXEC_UDF:
- federatedExecuteUDFCount.increment();
- break;
- default:
- break;
- }
- }
-
- public static void incFedAsyncPrefetchCount(long c) {
- fedAsyncPrefetchCount.add(c);
- }
public static void startCompileTimer() {
if( DMLScript.STATISTICS )
@@ -489,42 +224,17 @@ public class Statistics
public static void reset()
{
- hopRecompileTime.reset();
- hopRecompilePred.reset();
- hopRecompileSB.reset();
-
+ RecompileStatistics.reset();
+
funRecompiles.reset();
funRecompileTime.reset();
-
- codegenHopCompile.reset();
- codegenCPlanCompile.reset();
- codegenClassCompile.reset();
- codegenEnumAll.reset();
- codegenEnumAllP.reset();
- codegenEnumEval.reset();
- codegenEnumEvalP.reset();
- codegenCompileTime.reset();
- codegenClassCompileTime.reset();
- codegenOpCacheHits.reset();
- codegenOpCacheTotal.reset();
- codegenPlanCacheHits.reset();
- codegenPlanCacheTotal.reset();
-
- parforOptCount = 0;
- parforOptTime = 0;
- parforInitTime = 0;
- parforMergeTime = 0;
-
- sparkCtxCreateTime = 0;
- sparkBroadcast.reset();
- sparkBroadcastCount.reset();
- sparkAsyncPrefetchCount.reset();
- sparkAsyncBroadcastCount.reset();
- sparkParallelize.reset();
- sparkParallelizeCount.reset();
- sparkCollect.reset();
- sparkCollectCount.reset();
-
+
+ CodegenStatistics.reset();
+ ParForStatistics.reset();
+ ParamServStatistics.reset();
+ SparkStatistics.reset();
+ TransformStatistics.reset();
+
lTotalLix.reset();
lTotalLixUIP.reset();
lTotalUIPVar.reset();
@@ -538,26 +248,10 @@ public class Statistics
resetCPHeavyHitters();
GPUStatistics.reset();
- numNativeLibMatrixMultCalls.reset();
- numNativeSparseConv2dCalls.reset();
- numNativeSparseConv2dBwdDataCalls.reset();
- numNativeSparseConv2dBwdFilterCalls.reset();
- numNativeConv2dCalls.reset();
- numNativeConv2dBwdDataCalls.reset();
- numNativeConv2dBwdFilterCalls.reset();
- numNativeFailures.reset();
- nativeLibMatrixMultTime = 0;
- nativeConv2dTime = 0;
- nativeConv2dBwdFilterTime = 0;
- nativeConv2dBwdDataTime = 0;
- federatedReadCount.reset();
- federatedPutCount.reset();
- federatedGetCount.reset();
- federatedExecuteInstructionCount.reset();
- federatedExecuteUDFCount.reset();
- fedAsyncPrefetchCount.reset();
-
+ NativeStatistics.reset();
DMLCompressionStatistics.reset();
+
+ FederatedStatistics.reset();
}
public static void resetJITCompileTime(){
@@ -576,177 +270,6 @@ public class Statistics
_instStats.clear();
}
- public static void setSparkCtxCreateTime(long ns) {
- sparkCtxCreateTime = ns;
- }
-
- public static void accSparkParallelizeTime(long t) {
- sparkParallelize.add(t);
- }
-
- public static void incSparkParallelizeCount(long c) {
- sparkParallelizeCount.add(c);
- }
-
- public static void accSparkCollectTime(long t) {
- sparkCollect.add(t);
- sparkCollectCount.add(1);
- }
-
- public static long getSparkCollectCount(){
- return sparkCollectCount.longValue();
- }
-
- public static void accSparkBroadCastTime(long t) {
- sparkBroadcast.add(t);
- }
-
- public static void incSparkBroadcastCount(long c) {
- sparkBroadcastCount.add(c);
- }
-
- public static void incSparkAsyncPrefetchCount(long c) {
- sparkAsyncPrefetchCount.add(c);
- }
-
- public static void incSparkAsyncBroadcastCount(long c) {
- sparkAsyncBroadcastCount.add(c);
- }
-
- public static void incWorkerNumber() {
- psNumWorkers.increment();
- }
-
- public static void incWorkerNumber(long n) {
- psNumWorkers.add(n);
- }
-
- public static Timing getPSExecutionTimer() {
- return psExecutionTimer;
- }
-
- public static double getPSExecutionTime() {
- return psExecutionTime.doubleValue();
- }
-
- public static void accPSExecutionTime(long n) {
- psExecutionTime.add(n);
- }
-
- public static void accPSSetupTime(long t) {
- psSetupTime.add(t);
- }
-
- public static void accPSGradientComputeTime(long t) {
- psGradientComputeTime.add(t);
- }
-
- public static void accPSAggregationTime(long t) {
- psAggregationTime.add(t);
- }
-
- public static void accPSLocalModelUpdateTime(long t) {
- psLocalModelUpdateTime.add(t);
- }
-
- public static void accPSModelBroadcastTime(long t) {
- psModelBroadcastTime.add(t);
- }
-
- public static void accPSBatchIndexingTime(long t) {
- psBatchIndexTime.add(t);
- }
-
- public static void accPSRpcRequestTime(long t) {
- psRpcRequestTime.add(t);
- }
-
- public static double getPSValidationTime() {
- return psValidationTime.doubleValue();
- }
-
- public static void accPSValidationTime(long t) {
- psValidationTime.add(t);
- }
-
- public static void accFedPSDataPartitioningTime(long t) {
- fedPSDataPartitioningTime.add(t);
- }
-
- public static void accFedPSWorkerComputing(long t) {
- fedPSWorkerComputingTime.add(t);
- }
-
- public static void accFedPSGradientWeightingTime(long t) {
- fedPSGradientWeightingTime.add(t);
- }
-
- public static void accFedPSCommunicationTime(long t) { fedPSCommunicationTime.add(t);}
-
- public static void incTransformEncoderCount(long encoders){
- transformEncoderCount.add(encoders);
- }
-
- public static void incTransformRecodeApplyTime(long t){
- transformRecodeApplyTime.add(t);
- }
-
- public static void incTransformDummyCodeApplyTime(long t){
- transformDummyCodeApplyTime.add(t);
- }
-
- public static void incTransformBinningApplyTime(long t){
- transformBinningApplyTime.add(t);
- }
-
- public static void incTransformPassThroughApplyTime(long t){
- transformPassThroughApplyTime.add(t);
- }
-
- public static void incTransformFeatureHashingApplyTime(long t){
- transformFeatureHashingApplyTime.add(t);
- }
-
- public static void incTransformOmitApplyTime(long t) {
- transformOmitApplyTime.add(t);
- }
-
- public static void incTransformImputeApplyTime(long t) {
- transformImputeApplyTime.add(t);
- }
-
- public static void incTransformRecodeBuildTime(long t){
- transformRecodeBuildTime.add(t);
- }
-
- public static void incTransformBinningBuildTime(long t){
- transformBinningBuildTime.add(t);
- }
-
- public static void incTransformImputeBuildTime(long t) {
- transformImputeBuildTime.add(t);
- }
-
- public static void incTransformOutMatrixPreProcessingTime(long t){
- transformOutMatrixPreProcessingTime.add(t);
- }
-
- public static void incTransformOutMatrixPostProcessingTime(long t){
- transformOutMatrixPostProcessingTime.add(t);
- }
-
- public static long getTransformEncodeBuildTime(){
- return transformBinningBuildTime.longValue() + transformImputeBuildTime.longValue() +
- transformRecodeBuildTime.longValue();
- }
-
- public static long getTransformEncodeApplyTime(){
- return transformDummyCodeApplyTime.longValue() + transformBinningApplyTime.longValue() +
- transformFeatureHashingApplyTime.longValue() + transformPassThroughApplyTime.longValue() +
- transformRecodeApplyTime.longValue() + transformOmitApplyTime.longValue() +
- transformImputeApplyTime.longValue();
- }
-
public static String getCPHeavyHitterCode( Instruction inst )
{
String opcode = null;
@@ -1031,19 +554,7 @@ public class Statistics
return ret;
}
-
- public static long getHopRecompileTime(){
- return hopRecompileTime.longValue();
- }
-
- public static long getHopRecompiledPredDAGs(){
- return hopRecompilePred.longValue();
- }
-
- public static long getHopRecompiledSBDAGs(){
- return hopRecompileSB.longValue();
- }
-
+
public static long getFunRecompileTime(){
return funRecompileTime.longValue();
}
@@ -1051,34 +562,10 @@ public class Statistics
public static long getFunRecompiles(){
return funRecompiles.longValue();
}
-
- public static long getParforOptCount(){
- return parforOptCount;
- }
-
- public static long getParforOptTime(){
- return parforOptTime;
- }
-
- public static long getParforInitTime(){
- return parforInitTime;
- }
-
- public static long getParforMergeTime(){
- return parforMergeTime;
- }
public static long getNumPinnedObjects() { return maxNumPinnedObjects; }
public static double getSizeofPinnedObjects() { return maxSizeofPinnedObjects; }
-
- public static long getAsyncPrefetchCount() {
- return sparkAsyncPrefetchCount.longValue();
- }
-
- public static long getAsyncBroadcastCount() {
- return sparkAsyncBroadcastCount.longValue();
- }
/**
* Returns statistics of the DML program that was recently completed as a string
@@ -1126,18 +613,9 @@ public class Statistics
//show extended caching/compilation statistics
if( DMLScript.STATISTICS )
{
- if(NativeHelper.CURRENT_NATIVE_BLAS_STATE == NativeHelper.NativeBlasState.SUCCESSFULLY_LOADED_NATIVE_BLAS_AND_IN_USE) {
- String blas = NativeHelper.getCurrentBLAS();
- sb.append("Native " + blas + " calls (dense mult/conv/bwdF/bwdD):\t" + numNativeLibMatrixMultCalls.longValue() + "/" +
- numNativeConv2dCalls.longValue() + "/" + numNativeConv2dBwdFilterCalls.longValue()
- + "/" + numNativeConv2dBwdDataCalls.longValue() + ".\n");
- sb.append("Native " + blas + " calls (sparse conv/bwdF/bwdD):\t" +
- numNativeSparseConv2dCalls.longValue() + "/" + numNativeSparseConv2dBwdFilterCalls.longValue()
- + "/" + numNativeSparseConv2dBwdDataCalls.longValue() + ".\n");
- sb.append("Native " + blas + " times (dense mult/conv/bwdF/bwdD):\t" + String.format("%.3f", nativeLibMatrixMultTime*1e-9) + "/" +
- String.format("%.3f", nativeConv2dTime*1e-9) + "/" + String.format("%.3f", nativeConv2dBwdFilterTime*1e-9) + "/" +
- String.format("%.3f", nativeConv2dBwdDataTime*1e-9) + ".\n");
- }
+ if(NativeHelper.CURRENT_NATIVE_BLAS_STATE == NativeHelper.NativeBlasState.SUCCESSFULLY_LOADED_NATIVE_BLAS_AND_IN_USE)
+ sb.append(NativeStatistics.displayStatistics());
+
if(recomputeNNZTime != 0 || examSparsityTime != 0 || allocateDoubleArrTime != 0) {
sb.append("MatrixBlock times (recomputeNNZ/examSparsity/allocateDoubleArr):\t" + String.format("%.3f", recomputeNNZTime*1e-9) + "/" +
String.format("%.3f", examSparsityTime*1e-9) + "/" + String.format("%.3f", allocateDoubleArrTime*1e-9) + ".\n");
@@ -1148,8 +626,9 @@ public class Statistics
sb.append("Cache times (ACQr/m, RLS, EXP):\t" + CacheStatistics.displayTime() + " sec.\n");
if (DMLScript.JMLC_MEM_STATISTICS)
sb.append("Max size of live objects:\t" + byteCountToDisplaySize(getSizeofPinnedObjects()) + " (" + getNumPinnedObjects() + " total objects)" + "\n");
- sb.append("HOP DAGs recompiled (PRED, SB):\t" + getHopRecompiledPredDAGs() + "/" + getHopRecompiledSBDAGs() + ".\n");
- sb.append("HOP DAGs recompile time:\t" + String.format("%.3f", ((double)getHopRecompileTime())/1000000000) + " sec.\n");
+
+ sb.append(RecompileStatistics.displayStatistics());
+
if( getFunRecompiles()>0 ) {
sb.append("Functions recompiled:\t\t" + getFunRecompiles() + ".\n");
sb.append("Functions recompile time:\t" + String.format("%.3f", ((double)getFunRecompileTime())/1000000000) + " sec.\n");
@@ -1163,116 +642,21 @@ public class Statistics
sb.append("LinCache Computetime (S/M): \t" + LineageCacheStatistics.displayComputeTime() + " sec.\n");
sb.append("LinCache Rewrites: \t\t" + LineageCacheStatistics.displayRewrites() + ".\n");
}
- if( ConfigurationManager.isCodegenEnabled() ) {
- sb.append("Codegen compile (DAG,CP,JC):\t" + getCodegenDAGCompile() + "/"
- + getCodegenCPlanCompile() + "/" + getCodegenClassCompile() + ".\n");
- sb.append("Codegen enum (ALLt/p,EVALt/p):\t" + getCodegenEnumAll() + "/" +
- getCodegenEnumAllP() + "/" + getCodegenEnumEval() + "/" + getCodegenEnumEvalP() + ".\n");
- sb.append("Codegen compile times (DAG,JC):\t" + String.format("%.3f", (double)getCodegenCompileTime()/1000000000) + "/" +
- String.format("%.3f", (double)getCodegenClassCompileTime()/1000000000) + " sec.\n");
- sb.append("Codegen enum plan cache hits:\t" + getCodegenPlanCacheHits() + "/" + getCodegenPlanCacheTotal() + ".\n");
- sb.append("Codegen op plan cache hits:\t" + getCodegenOpCacheHits() + "/" + getCodegenOpCacheTotal() + ".\n");
- }
- if( OptimizerUtils.isSparkExecutionMode() ){
- String lazy = SparkExecutionContext.isLazySparkContextCreation() ? "(lazy)" : "(eager)";
- sb.append("Spark ctx create time "+lazy+":\t"+
- String.format("%.3f", sparkCtxCreateTime*1e-9) + " sec.\n" ); // nanoSec --> sec
- sb.append("Spark trans counts (par,bc,col):" +
- String.format("%d/%d/%d.\n", sparkParallelizeCount.longValue(),
- sparkBroadcastCount.longValue(), sparkCollectCount.longValue()));
- sb.append("Spark trans times (par,bc,col):\t" +
- String.format("%.3f/%.3f/%.3f secs.\n",
- sparkParallelize.longValue()*1e-9,
- sparkBroadcast.longValue()*1e-9,
- sparkCollect.longValue()*1e-9));
- if (OptimizerUtils.ASYNC_TRIGGER_RDD_OPERATIONS)
- sb.append("Spark async. count (pf,bc): \t" +
- String.format("%d/%d.\n", getAsyncPrefetchCount(), getAsyncBroadcastCount()));
- }
- if (psNumWorkers.longValue() > 0) {
- sb.append(String.format("Paramserv total execution time:\t%.3f secs.\n", psExecutionTime.doubleValue() / 1000));
- sb.append(String.format("Paramserv total num workers:\t%d.\n", psNumWorkers.longValue()));
- sb.append(String.format("Paramserv setup time:\t\t%.3f secs.\n", psSetupTime.doubleValue() / 1000));
-
- if(fedPSDataPartitioningTime.doubleValue() > 0) { //if data partitioning happens this is the federated case
- sb.append(String.format("PS fed data partitioning time:\t%.3f secs.\n", fedPSDataPartitioningTime.doubleValue() / 1000));
- sb.append(String.format("PS fed comm time (cum):\t\t%.3f secs.\n", fedPSCommunicationTime.doubleValue() / 1000));
- sb.append(String.format("PS fed worker comp time (cum):\t%.3f secs.\n", fedPSWorkerComputingTime.doubleValue() / 1000));
- sb.append(String.format("PS fed grad. weigh. time (cum):\t%.3f secs.\n", fedPSGradientWeightingTime.doubleValue() / 1000));
- sb.append(String.format("PS fed global model agg time:\t%.3f secs.\n", psAggregationTime.doubleValue() / 1000));
- }
- else {
- sb.append(String.format("Paramserv grad compute time:\t%.3f secs.\n", psGradientComputeTime.doubleValue() / 1000));
- sb.append(String.format("Paramserv model update time:\t%.3f/%.3f secs.\n",
- psLocalModelUpdateTime.doubleValue() / 1000, psAggregationTime.doubleValue() / 1000));
- sb.append(String.format("Paramserv model broadcast time:\t%.3f secs.\n", psModelBroadcastTime.doubleValue() / 1000));
- sb.append(String.format("Paramserv batch slice time:\t%.3f secs.\n", psBatchIndexTime.doubleValue() / 1000));
- sb.append(String.format("Paramserv RPC request time:\t%.3f secs.\n", psRpcRequestTime.doubleValue() / 1000));
- }
- sb.append(String.format("Paramserv valdiation time:\t%.3f secs.\n", psValidationTime.doubleValue() / 1000));
- }
- if( parforOptCount>0 ){
- sb.append("ParFor loops optimized:\t\t" + getParforOptCount() + ".\n");
- sb.append("ParFor optimize time:\t\t" + String.format("%.3f", ((double)getParforOptTime())/1000) + " sec.\n");
- sb.append("ParFor initialize time:\t\t" + String.format("%.3f", ((double)getParforInitTime())/1000) + " sec.\n");
- sb.append("ParFor result merge time:\t" + String.format("%.3f", ((double)getParforMergeTime())/1000) + " sec.\n");
- sb.append("ParFor total update in-place:\t" + lTotalUIPVar + "/" + lTotalLixUIP + "/" + lTotalLix + "\n");
- }
- if( federatedReadCount.longValue() > 0){
- sb.append("Federated I/O (Read, Put, Get):\t" +
- federatedReadCount.longValue() + "/" +
- federatedPutCount.longValue() + "/" +
- federatedGetCount.longValue() + ".\n");
- sb.append("Federated Execute (Inst, UDF):\t" +
- federatedExecuteInstructionCount.longValue() + "/" +
- federatedExecuteUDFCount.longValue() + ".\n");
- sb.append("Federated prefetch count:\t" +
- fedAsyncPrefetchCount.longValue() + ".\n");
- }
- if( transformEncoderCount.longValue() > 0) {
- //TODO: Cleanup and condense
- sb.append("TransformEncode num. encoders:\t").append(transformEncoderCount.longValue()).append("\n");
- sb.append("TransformEncode build time:\t").append(String.format("%.3f",
- getTransformEncodeBuildTime()*1e-9)).append(" sec.\n");
- if(transformRecodeBuildTime.longValue() > 0)
- sb.append("\tRecode build time:\t").append(String.format("%.3f",
- transformRecodeBuildTime.longValue()*1e-9)).append(" sec.\n");
- if(transformBinningBuildTime.longValue() > 0)
- sb.append("\tBinning build time:\t").append(String.format("%.3f",
- transformBinningBuildTime.longValue()*1e-9)).append(" sec.\n");
- if(transformImputeBuildTime.longValue() > 0)
- sb.append("\tImpute build time:\t").append(String.format("%.3f",
- transformImputeBuildTime.longValue()*1e-9)).append(" sec.\n");
-
- sb.append("TransformEncode apply time:\t").append(String.format("%.3f",
- getTransformEncodeApplyTime()*1e-9)).append(" sec.\n");
- if(transformRecodeApplyTime.longValue() > 0)
- sb.append("\tRecode apply time:\t").append(String.format("%.3f",
- transformRecodeApplyTime.longValue()*1e-9)).append(" sec.\n");
- if(transformBinningApplyTime.longValue() > 0)
- sb.append("\tBinning apply time:\t").append(String.format("%.3f",
- transformBinningApplyTime.longValue()*1e-9)).append(" sec.\n");
- if(transformDummyCodeApplyTime.longValue() > 0)
- sb.append("\tDummyCode apply time:\t").append(String.format("%.3f",
- transformDummyCodeApplyTime.longValue()*1e-9)).append(" sec.\n");
- if(transformFeatureHashingApplyTime.longValue() > 0)
- sb.append("\tHashing apply time:\t").append(String.format("%.3f",
- transformFeatureHashingApplyTime.longValue()*1e-9)).append(" sec.\n");
- if(transformPassThroughApplyTime.longValue() > 0)
- sb.append("\tPassThrough apply time:\t").append(String.format("%.3f",
- transformPassThroughApplyTime.longValue()*1e-9)).append(" sec.\n");
- if(transformOmitApplyTime.longValue() > 0)
- sb.append("\tOmit apply time:\t").append(String.format("%.3f",
- transformOmitApplyTime.longValue()*1e-9)).append(" sec.\n");
- if(transformImputeApplyTime.longValue() > 0)
- sb.append("\tImpute apply time:\t").append(String.format("%.3f",
- transformImputeApplyTime.longValue()*1e-9)).append(" sec.\n");
-
- sb.append("TransformEncode PreProc. time:\t").append(String.format("%.3f",
- transformOutMatrixPreProcessingTime.longValue()*1e-9)).append(" sec.\n");
- sb.append("TransformEncode PostProc. time:\t").append(String.format("%.3f",
- transformOutMatrixPostProcessingTime.longValue()*1e-9)).append(" sec.\n");
- }
+
+ if( ConfigurationManager.isCodegenEnabled() )
+ sb.append(CodegenStatistics.displayStatistics());
+
+ if( OptimizerUtils.isSparkExecutionMode() )
+ sb.append(SparkStatistics.displayStatistics());
+
+ sb.append(ParamServStatistics.displayStatistics());
+
+ sb.append(ParForStatistics.displayStatistics());
+
+ sb.append(FederatedStatistics.displayFedIOExecStatistics());
+ sb.append(FederatedStatistics.displayFedLookupTableStats());
+
+ sb.append(TransformStatistics.displayStatistics());
if(ConfigurationManager.isCompressionEnabled()){
DMLCompressionStatistics.display(sb);
@@ -1289,7 +673,7 @@ public class Statistics
if(DMLScript.FED_STATISTICS) {
sb.append("\n");
- sb.append(FederatedStatistics.displayFedStatistics(DMLScript.FED_STATISTICS_COUNT));
+ sb.append(FederatedStatistics.displayStatistics(DMLScript.FED_STATISTICS_COUNT));
}
return sb.toString();
diff --git a/src/main/java/org/apache/sysds/utils/stats/CodegenStatistics.java b/src/main/java/org/apache/sysds/utils/stats/CodegenStatistics.java
new file mode 100644
index 0000000..e84b6b9
--- /dev/null
+++ b/src/main/java/org/apache/sysds/utils/stats/CodegenStatistics.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.utils.stats;
+
+import java.util.concurrent.atomic.LongAdder;
+
+public class CodegenStatistics {
+ private static final LongAdder compileTime = new LongAdder(); //in nano
+ private static final LongAdder classCompileTime = new LongAdder(); //in nano
+ private static final LongAdder hopCompile = new LongAdder(); //count
+ private static final LongAdder cPlanCompile = new LongAdder(); //count
+ private static final LongAdder classCompile = new LongAdder(); //count
+ private static final LongAdder enumAll = new LongAdder(); //count
+ private static final LongAdder enumAllP = new LongAdder(); //count
+ private static final LongAdder enumEval = new LongAdder(); //count
+ private static final LongAdder enumEvalP = new LongAdder(); //count
+ private static final LongAdder opCacheHits = new LongAdder(); //count
+ private static final LongAdder opCacheTotal = new LongAdder(); //count
+ private static final LongAdder planCacheHits = new LongAdder(); //count
+ private static final LongAdder planCacheTotal = new LongAdder(); //count
+
+
+ public static void incrementDAGCompile() {
+ hopCompile.increment();
+ }
+
+ public static void incrementCPlanCompile(long delta) {
+ cPlanCompile.add(delta);
+ }
+
+ public static void incrementEnumAll(long delta) {
+ enumAll.add(delta);
+ }
+ public static void incrementEnumAllP(long delta) {
+ enumAllP.add(delta);
+ }
+ public static void incrementEnumEval(long delta) {
+ enumEval.add(delta);
+ }
+ public static void incrementEnumEvalP(long delta) {
+ enumEvalP.add(delta);
+ }
+
+ public static void incrementClassCompile() {
+ classCompile.increment();
+ }
+
+ public static void incrementCompileTime(long delta) {
+ compileTime.add(delta);
+ }
+
+ public static void incrementClassCompileTime(long delta) {
+ classCompileTime.add(delta);
+ }
+
+ public static void incrementOpCacheHits() {
+ opCacheHits.increment();
+ }
+
+ public static void incrementOpCacheTotal() {
+ opCacheTotal.increment();
+ }
+
+ public static void incrementPlanCacheHits() {
+ planCacheHits.increment();
+ }
+
+ public static void incrementPlanCacheTotal() {
+ planCacheTotal.increment();
+ }
+
+ public static long getDAGCompile() {
+ return hopCompile.longValue();
+ }
+
+ public static long getCPlanCompile() {
+ return cPlanCompile.longValue();
+ }
+
+ public static long getEnumAll() {
+ return enumAll.longValue();
+ }
+
+ public static long getEnumAllP() {
+ return enumAllP.longValue();
+ }
+
+ public static long getEnumEval() {
+ return enumEval.longValue();
+ }
+
+ public static long getEnumEvalP() {
+ return enumEvalP.longValue();
+ }
+
+ public static long getClassCompile() {
+ return classCompile.longValue();
+ }
+
+ public static long getCompileTime() {
+ return compileTime.longValue();
+ }
+
+ public static long getClassCompileTime() {
+ return classCompileTime.longValue();
+ }
+
+ public static long getOpCacheHits() {
+ return opCacheHits.longValue();
+ }
+
+ public static long getOpCacheTotal() {
+ return opCacheTotal.longValue();
+ }
+
+ public static long getPlanCacheHits() {
+ return planCacheHits.longValue();
+ }
+
+ public static long getPlanCacheTotal() {
+ return planCacheTotal.longValue();
+ }
+
+ public static void reset() {
+ hopCompile.reset();
+ cPlanCompile.reset();
+ classCompile.reset();
+ enumAll.reset();
+ enumAllP.reset();
+ enumEval.reset();
+ enumEvalP.reset();
+ compileTime.reset();
+ classCompileTime.reset();
+ opCacheHits.reset();
+ opCacheTotal.reset();
+ planCacheHits.reset();
+ planCacheTotal.reset();
+ }
+
+ public static String displayStatistics() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Codegen compile (DAG,CP,JC):\t" + getDAGCompile() + "/"
+ + getCPlanCompile() + "/" + getClassCompile() + ".\n");
+ sb.append("Codegen enum (ALLt/p,EVALt/p):\t" + getEnumAll() + "/" +
+ getEnumAllP() + "/" + getEnumEval() + "/" + getEnumEvalP() + ".\n");
+ sb.append("Codegen compile times (DAG,JC):\t" + String.format("%.3f", (double)getCompileTime()/1000000000) + "/" +
+ String.format("%.3f", (double)getClassCompileTime()/1000000000) + " sec.\n");
+ sb.append("Codegen enum plan cache hits:\t" + getPlanCacheHits() + "/" + getPlanCacheTotal() + ".\n");
+ sb.append("Codegen op plan cache hits:\t" + getOpCacheHits() + "/" + getOpCacheTotal() + ".\n");
+ return sb.toString();
+ }
+}
diff --git a/src/main/java/org/apache/sysds/utils/stats/NativeStatistics.java b/src/main/java/org/apache/sysds/utils/stats/NativeStatistics.java
new file mode 100644
index 0000000..c0632cf
--- /dev/null
+++ b/src/main/java/org/apache/sysds/utils/stats/NativeStatistics.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.utils.stats;
+
+import java.util.concurrent.atomic.LongAdder;
+
+import org.apache.sysds.utils.NativeHelper;
+
+public class NativeStatistics {
+ private static LongAdder numFailures = new LongAdder();
+ private static LongAdder numLibMatrixMultCalls = new LongAdder();
+ private static LongAdder libMatrixMultTime = new LongAdder();
+ private static LongAdder numConv2dCalls = new LongAdder();
+ private static LongAdder conv2dTime = new LongAdder();
+ private static LongAdder numConv2dBwdDataCalls = new LongAdder();
+ private static LongAdder conv2dBwdDataTime = new LongAdder();
+ private static LongAdder numConv2dBwdFilterCalls = new LongAdder();
+ private static LongAdder conv2dBwdFilterTime = new LongAdder();
+ private static LongAdder numSparseConv2dCalls = new LongAdder();
+ private static LongAdder numSparseConv2dBwdFilterCalls = new LongAdder();
+ private static LongAdder numSparseConv2dBwdDataCalls = new LongAdder();
+
+ public static void incrementFailuresCounter() {
+ numFailures.increment();
+ // This is very rare and am not sure it is possible at all. Our initial experiments never encountered this case.
+ // Note: all the native calls have a fallback to Java; so if the user wants she can recompile SystemDS by
+ // commenting this exception and everything should work fine.
+ throw new RuntimeException("Unexpected ERROR: OOM caused during JNI transfer. Please disable native BLAS by setting enviroment variable: SYSTEMDS_BLAS=none");
+ }
+
+ public static void incrementNumLibMatrixMultCalls() {
+ numLibMatrixMultCalls.increment();
+ }
+
+ public static void incrementLibMatrixMultTime(long time) {
+ libMatrixMultTime.add(time);
+ }
+
+ public static void incrementNumConv2dCalls() {
+ numConv2dCalls.increment();
+ }
+
+ public static void incrementConv2dTime(long time) {
+ conv2dTime.add(time);
+ }
+
+ public static void incrementNumConv2dBwdDataCalls() {
+ numConv2dBwdDataCalls.increment();
+ }
+
+ public static void incrementConv2dBwdDataTime(long time) {
+ conv2dBwdDataTime.add(time);
+ }
+
+ public static void incrementNumConv2dBwdFilterCalls() {
+ numConv2dBwdFilterCalls.increment();
+ }
+
+ public static void incrementConv2dBwdFilterTime(long time) {
+ conv2dBwdFilterTime.add(time);
+ }
+
+ public static void incrementNumSparseConv2dCalls() {
+ numSparseConv2dCalls.increment();
+ }
+
+ public static void incrementNumSparseConv2dBwdFilterCalls() {
+ numSparseConv2dBwdFilterCalls.increment();
+ }
+
+ public static void incrementNumSparseConv2dBwdDataCalls() {
+ numSparseConv2dBwdDataCalls.increment();
+ }
+
+ public static void reset() {
+ numLibMatrixMultCalls.reset();
+ numSparseConv2dCalls.reset();
+ numSparseConv2dBwdDataCalls.reset();
+ numSparseConv2dBwdFilterCalls.reset();
+ numConv2dCalls.reset();
+ numConv2dBwdDataCalls.reset();
+ numConv2dBwdFilterCalls.reset();
+ numFailures.reset();
+ libMatrixMultTime.reset();
+ conv2dTime.reset();
+ conv2dBwdFilterTime.reset();
+ conv2dBwdDataTime.reset();
+ }
+
+ public static String displayStatistics() {
+ StringBuilder sb = new StringBuilder();
+ String blas = NativeHelper.getCurrentBLAS();
+ sb.append("Native " + blas + " calls (dense mult/conv/bwdF/bwdD):\t" + numLibMatrixMultCalls.longValue() + "/"
+ + numConv2dCalls.longValue() + "/" + numConv2dBwdFilterCalls.longValue()
+ + "/" + numConv2dBwdDataCalls.longValue() + ".\n");
+ sb.append("Native " + blas + " calls (sparse conv/bwdF/bwdD):\t"
+ + numSparseConv2dCalls.longValue() + "/" + numSparseConv2dBwdFilterCalls.longValue()
+ + "/" + numSparseConv2dBwdDataCalls.longValue() + ".\n");
+ sb.append("Native " + blas + " times (dense mult/conv/bwdF/bwdD):\t"
+ + String.format("%.3f", libMatrixMultTime.longValue()*1e-9) + "/"
+ + String.format("%.3f", conv2dTime.longValue()*1e-9) + "/"
+ + String.format("%.3f", conv2dBwdFilterTime.longValue()*1e-9) + "/"
+ + String.format("%.3f", conv2dBwdDataTime.longValue()*1e-9) + ".\n");
+ return sb.toString();
+ }
+}
diff --git a/src/main/java/org/apache/sysds/utils/stats/ParForStatistics.java b/src/main/java/org/apache/sysds/utils/stats/ParForStatistics.java
new file mode 100644
index 0000000..b4f5b58
--- /dev/null
+++ b/src/main/java/org/apache/sysds/utils/stats/ParForStatistics.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.utils.stats;
+
+import java.util.concurrent.atomic.LongAdder;
+
+import org.apache.sysds.utils.Statistics;
+
+public class ParForStatistics {
+ //PARFOR optimization stats (low frequency updates)
+ private static final LongAdder optTime = new LongAdder(); //in milli sec
+ private static final LongAdder optCount = new LongAdder(); //count
+ private static final LongAdder initTime = new LongAdder(); //in milli sec
+ private static final LongAdder mergeTime = new LongAdder(); //in milli sec
+
+ public static synchronized void incrementOptimCount(){
+ optCount.increment();
+ }
+
+ public static synchronized void incrementOptimTime( long time ) {
+ optTime.add(time);
+ }
+
+ public static synchronized void incrementInitTime( long time ) {
+ initTime.add(time);
+ }
+
+ public static synchronized void incrementMergeTime( long time ) {
+ mergeTime.add(time);
+ }
+
+ public static long getOptCount(){
+ return optCount.longValue();
+ }
+
+ public static long getOptTime(){
+ return optTime.longValue();
+ }
+
+ public static long getInitTime(){
+ return initTime.longValue();
+ }
+
+ public static long getMergeTime(){
+ return mergeTime.longValue();
+ }
+
+ public static void reset() {
+ optCount.reset();
+ optTime.reset();
+ initTime.reset();
+ mergeTime.reset();
+ }
+
+ public static String displayStatistics() {
+ if( optCount.longValue() > 0 ){
+ StringBuilder sb = new StringBuilder();
+ sb.append("ParFor loops optimized:\t\t" + getOptCount() + ".\n");
+ sb.append("ParFor optimize time:\t\t" + String.format("%.3f", ((double)getOptTime())/1000) + " sec.\n");
+ sb.append("ParFor initialize time:\t\t" + String.format("%.3f", ((double)getInitTime())/1000) + " sec.\n");
+ sb.append("ParFor result merge time:\t" + String.format("%.3f", ((double)getMergeTime())/1000) + " sec.\n");
+ sb.append("ParFor total update in-place:\t" + Statistics.getTotalUIPVar() + "/"
+ + Statistics.getTotalLixUIP() + "/" + Statistics.getTotalLix() + "\n");
+ return sb.toString();
+ }
+ return "";
+ }
+}
diff --git a/src/main/java/org/apache/sysds/utils/stats/ParamServStatistics.java b/src/main/java/org/apache/sysds/utils/stats/ParamServStatistics.java
new file mode 100644
index 0000000..0d97bfd
--- /dev/null
+++ b/src/main/java/org/apache/sysds/utils/stats/ParamServStatistics.java
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.utils.stats;
+
+import java.util.concurrent.atomic.LongAdder;
+
+import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
+
+public class ParamServStatistics {
+ // Paramserv function stats (time is in milli sec)
+ private static final Timing executionTimer = new Timing(false);
+ private static final LongAdder executionTime = new LongAdder();
+ private static final LongAdder numWorkers = new LongAdder();
+ private static final LongAdder setupTime = new LongAdder();
+ private static final LongAdder gradientComputeTime = new LongAdder();
+ private static final LongAdder aggregationTime = new LongAdder();
+ private static final LongAdder localModelUpdateTime = new LongAdder();
+ private static final LongAdder modelBroadcastTime = new LongAdder();
+ private static final LongAdder batchIndexTime = new LongAdder();
+ private static final LongAdder rpcRequestTime = new LongAdder();
+ private static final LongAdder validationTime = new LongAdder();
+ // Federated parameter server specifics (time is in milli sec)
+ private static final LongAdder fedDataPartitioningTime = new LongAdder();
+ private static final LongAdder fedWorkerComputingTime = new LongAdder();
+ private static final LongAdder fedGradientWeightingTime = new LongAdder();
+ private static final LongAdder fedCommunicationTime = new LongAdder();
+
+ public static void incWorkerNumber() {
+ numWorkers.increment();
+ }
+
+ public static void incWorkerNumber(long n) {
+ numWorkers.add(n);
+ }
+
+ public static Timing getExecutionTimer() {
+ return executionTimer;
+ }
+
+ public static double getExecutionTime() {
+ return executionTime.doubleValue();
+ }
+
+ public static void accExecutionTime(long n) {
+ executionTime.add(n);
+ }
+
+ public static void accSetupTime(long t) {
+ setupTime.add(t);
+ }
+
+ public static void accGradientComputeTime(long t) {
+ gradientComputeTime.add(t);
+ }
+
+ public static void accAggregationTime(long t) {
+ aggregationTime.add(t);
+ }
+
+ public static void accLocalModelUpdateTime(long t) {
+ localModelUpdateTime.add(t);
+ }
+
+ public static void accModelBroadcastTime(long t) {
+ modelBroadcastTime.add(t);
+ }
+
+ public static void accBatchIndexingTime(long t) {
+ batchIndexTime.add(t);
+ }
+
+ public static void accRpcRequestTime(long t) {
+ rpcRequestTime.add(t);
+ }
+
+ public static double getValidationTime() {
+ return validationTime.doubleValue();
+ }
+
+ public static void accValidationTime(long t) {
+ validationTime.add(t);
+ }
+
+ public static long getFedDataPartitioningTime() {
+ return fedDataPartitioningTime.longValue();
+ }
+
+ public static void accFedDataPartitioningTime(long t) {
+ fedDataPartitioningTime.add(t);
+ }
+
+ public static void accFedWorkerComputing(long t) {
+ fedWorkerComputingTime.add(t);
+ }
+
+ public static void accFedGradientWeightingTime(long t) {
+ fedGradientWeightingTime.add(t);
+ }
+
+ public static void accFedCommunicationTime(long t) {
+ fedCommunicationTime.add(t);
+ }
+
+ public static void reset() {
+ executionTime.reset();
+ numWorkers.reset();
+ setupTime.reset();
+ gradientComputeTime.reset();
+ aggregationTime.reset();
+ localModelUpdateTime.reset();
+ modelBroadcastTime.reset();
+ batchIndexTime.reset();
+ rpcRequestTime.reset();
+ validationTime.reset();
+ fedDataPartitioningTime.reset();
+ fedWorkerComputingTime.reset();
+ fedGradientWeightingTime.reset();
+ fedCommunicationTime.reset();
+ }
+
+ public static String displayStatistics() {
+ if (numWorkers.longValue() > 0) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(String.format("Paramserv total execution time:\t%.3f secs.\n", executionTime.doubleValue() / 1000));
+ sb.append(String.format("Paramserv total num workers:\t%d.\n", numWorkers.longValue()));
+ sb.append(String.format("Paramserv setup time:\t\t%.3f secs.\n", setupTime.doubleValue() / 1000));
+
+ if(fedDataPartitioningTime.longValue() > 0) { //if data partitioning happens this is the federated case
+ sb.append(displayFedPSStatistics());
+ sb.append(String.format("PS fed global model agg time:\t%.3f secs.\n", aggregationTime.doubleValue() / 1000));
+ }
+ else {
+ sb.append(String.format("Paramserv grad compute time:\t%.3f secs.\n", gradientComputeTime.doubleValue() / 1000));
+ sb.append(String.format("Paramserv model update time:\t%.3f/%.3f secs.\n",
+ localModelUpdateTime.doubleValue() / 1000, aggregationTime.doubleValue() / 1000));
+ sb.append(String.format("Paramserv model broadcast time:\t%.3f secs.\n", modelBroadcastTime.doubleValue() / 1000));
+ sb.append(String.format("Paramserv batch slice time:\t%.3f secs.\n", batchIndexTime.doubleValue() / 1000));
+ sb.append(String.format("Paramserv RPC request time:\t%.3f secs.\n", rpcRequestTime.doubleValue() / 1000));
+ }
+ sb.append(String.format("Paramserv valdiation time:\t%.3f secs.\n", validationTime.doubleValue() / 1000));
+ return sb.toString();
+ }
+ return "";
+ }
+
+ private static String displayFedPSStatistics() {
+ StringBuilder sb = new StringBuilder();
+ sb.append(String.format("PS fed data partitioning time:\t%.3f secs.\n", fedDataPartitioningTime.doubleValue() / 1000));
+ sb.append(String.format("PS fed comm time (cum):\t\t%.3f secs.\n", fedCommunicationTime.doubleValue() / 1000));
+ sb.append(String.format("PS fed worker comp time (cum):\t%.3f secs.\n", fedWorkerComputingTime.doubleValue() / 1000));
+ sb.append(String.format("PS fed grad. weigh. time (cum):\t%.3f secs.\n", fedGradientWeightingTime.doubleValue() / 1000));
+ return sb.toString();
+ }
+}
diff --git a/src/main/java/org/apache/sysds/utils/stats/RecompileStatistics.java b/src/main/java/org/apache/sysds/utils/stats/RecompileStatistics.java
new file mode 100644
index 0000000..53c655b
--- /dev/null
+++ b/src/main/java/org/apache/sysds/utils/stats/RecompileStatistics.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.utils.stats;
+
+import java.util.concurrent.atomic.LongAdder;
+
+public class RecompileStatistics {
+ //HOP DAG recompile stats (potentially high update frequency)
+ private static final LongAdder recompileTime = new LongAdder(); //in nano sec
+ private static final LongAdder recompilePred = new LongAdder(); //count
+ private static final LongAdder recompileSB = new LongAdder(); //count
+
+
+ public static void incrementRecompileTime( long delta ) {
+ recompileTime.add(delta);
+ }
+
+ public static void incrementRecompilePred() {
+ recompilePred.increment();
+ }
+
+ public static void incrementRecompilePred(long delta) {
+ recompilePred.add(delta);
+ }
+
+ public static void incrementRecompileSB() {
+ recompileSB.increment();
+ }
+
+ public static void incrementRecompileSB(long delta) {
+ recompileSB.add(delta);
+ }
+
+ public static long getRecompileTime(){
+ return recompileTime.longValue();
+ }
+
+ public static long getRecompiledPredDAGs(){
+ return recompilePred.longValue();
+ }
+
+ public static long getRecompiledSBDAGs(){
+ return recompileSB.longValue();
+ }
+
+ public static void reset() {
+ recompileTime.reset();
+ recompilePred.reset();
+ recompileSB.reset();
+ }
+
+ public static String displayStatistics() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("HOP DAGs recompiled (PRED, SB):\t" + getRecompiledPredDAGs() + "/" + getRecompiledSBDAGs() + ".\n");
+ sb.append("HOP DAGs recompile time:\t" + String.format("%.3f", ((double)getRecompileTime())/1000000000) + " sec.\n");
+ return sb.toString();
+ }
+}
diff --git a/src/main/java/org/apache/sysds/utils/stats/SparkStatistics.java b/src/main/java/org/apache/sysds/utils/stats/SparkStatistics.java
new file mode 100644
index 0000000..5263dbd
--- /dev/null
+++ b/src/main/java/org/apache/sysds/utils/stats/SparkStatistics.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.utils.stats;
+
+import java.util.concurrent.atomic.LongAdder;
+
+import org.apache.sysds.hops.OptimizerUtils;
+import org.apache.sysds.runtime.controlprogram.context.SparkExecutionContext;
+
+public class SparkStatistics {
+ private static long ctxCreateTime = 0;
+ private static final LongAdder parallelizeTime = new LongAdder();
+ private static final LongAdder parallelizeCount = new LongAdder();
+ private static final LongAdder collectTime = new LongAdder();
+ private static final LongAdder collectCount = new LongAdder();
+ private static final LongAdder broadcastTime = new LongAdder();
+ private static final LongAdder broadcastCount = new LongAdder();
+ private static final LongAdder asyncPrefetchCount = new LongAdder();
+ private static final LongAdder asyncBroadcastCount = new LongAdder();
+
+ public static boolean createdSparkContext() {
+ return ctxCreateTime > 0;
+ }
+
+ public static void setCtxCreateTime(long ns) {
+ ctxCreateTime = ns;
+ }
+
+ public static void accParallelizeTime(long t) {
+ parallelizeTime.add(t);
+ }
+
+ public static void incParallelizeCount(long c) {
+ parallelizeCount.add(c);
+ }
+
+ public static void accCollectTime(long t) {
+ collectTime.add(t);
+ incCollectCount(1);
+ }
+
+ private static void incCollectCount(long c) {
+ collectCount.add(c);
+ }
+
+ public static void accBroadCastTime(long t) {
+ broadcastTime.add(t);
+ }
+
+ public static void incBroadcastCount(long c) {
+ broadcastCount.add(c);
+ }
+
+ public static void incAsyncPrefetchCount(long c) {
+ asyncPrefetchCount.add(c);
+ }
+
+ public static void incAsyncBroadcastCount(long c) {
+ asyncBroadcastCount.add(c);
+ }
+
+ public static long getSparkCollectCount() {
+ return collectCount.longValue();
+ }
+
+ public static long getAsyncPrefetchCount() {
+ return asyncPrefetchCount.longValue();
+ }
+
+ public static long getAsyncBroadcastCount() {
+ return asyncBroadcastCount.longValue();
+ }
+
+ public static void reset() {
+ ctxCreateTime = 0;
+ parallelizeTime.reset();
+ parallelizeCount.reset();
+ broadcastTime.reset();
+ broadcastCount.reset();
+ collectTime.reset();
+ collectCount.reset();
+ asyncPrefetchCount.reset();
+ asyncBroadcastCount.reset();
+ }
+
+ public static String displayStatistics() {
+ StringBuilder sb = new StringBuilder();
+ String lazy = SparkExecutionContext.isLazySparkContextCreation() ? "(lazy)" : "(eager)";
+ sb.append("Spark ctx create time "+lazy+":\t"+
+ String.format("%.3f", ctxCreateTime*1e-9) + " sec.\n" ); // nanoSec --> sec
+ sb.append("Spark trans counts (par,bc,col):" +
+ String.format("%d/%d/%d.\n", parallelizeCount.longValue(),
+ broadcastCount.longValue(), collectCount.longValue()));
+ sb.append("Spark trans times (par,bc,col):\t" +
+ String.format("%.3f/%.3f/%.3f secs.\n",
+ parallelizeTime.longValue()*1e-9,
+ broadcastTime.longValue()*1e-9,
+ collectTime.longValue()*1e-9));
+ if (OptimizerUtils.ASYNC_TRIGGER_RDD_OPERATIONS)
+ sb.append("Spark async. count (pf,bc): \t" +
+ String.format("%d/%d.\n", getAsyncPrefetchCount(), getAsyncBroadcastCount()));
+ return sb.toString();
+ }
+}
diff --git a/src/main/java/org/apache/sysds/utils/stats/TransformStatistics.java b/src/main/java/org/apache/sysds/utils/stats/TransformStatistics.java
new file mode 100644
index 0000000..d572b42
--- /dev/null
+++ b/src/main/java/org/apache/sysds/utils/stats/TransformStatistics.java
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.utils.stats;
+
+import java.util.concurrent.atomic.LongAdder;
+
+public class TransformStatistics {
+ private static final LongAdder encoderCount = new LongAdder();
+
+ //private static final LongAdder buildTime = new LongAdder();
+ private static final LongAdder recodeBuildTime = new LongAdder();
+ private static final LongAdder binningBuildTime = new LongAdder();
+ private static final LongAdder imputeBuildTime = new LongAdder();
+
+ //private static final LongAdder applyTime = new LongAdder();
+ private static final LongAdder recodeApplyTime = new LongAdder();
+ private static final LongAdder dummyCodeApplyTime = new LongAdder();
+ private static final LongAdder passThroughApplyTime = new LongAdder();
+ private static final LongAdder featureHashingApplyTime = new LongAdder();
+ private static final LongAdder binningApplyTime = new LongAdder();
+ private static final LongAdder omitApplyTime = new LongAdder();
+ private static final LongAdder imputeApplyTime = new LongAdder();
+
+ private static final LongAdder outMatrixPreProcessingTime = new LongAdder();
+ private static final LongAdder outMatrixPostProcessingTime = new LongAdder();
+
+ public static void incEncoderCount(long encoders) {
+ encoderCount.add(encoders);
+ }
+
+ public static void incRecodeApplyTime(long t) {
+ recodeApplyTime.add(t);
+ }
+
+ public static void incDummyCodeApplyTime(long t) {
+ dummyCodeApplyTime.add(t);
+ }
+
+ public static void incBinningApplyTime(long t) {
+ binningApplyTime.add(t);
+ }
+
+ public static void incPassThroughApplyTime(long t) {
+ passThroughApplyTime.add(t);
+ }
+
+ public static void incFeatureHashingApplyTime(long t) {
+ featureHashingApplyTime.add(t);
+ }
+
+ public static void incOmitApplyTime(long t) {
+ omitApplyTime.add(t);
+ }
+
+ public static void incImputeApplyTime(long t) {
+ imputeApplyTime.add(t);
+ }
+
+ public static void incRecodeBuildTime(long t) {
+ recodeBuildTime.add(t);
+ }
+
+ public static void incBinningBuildTime(long t) {
+ binningBuildTime.add(t);
+ }
+
+ public static void incImputeBuildTime(long t) {
+ imputeBuildTime.add(t);
+ }
+
+ public static void incOutMatrixPreProcessingTime(long t) {
+ outMatrixPreProcessingTime.add(t);
+ }
+
+ public static void incOutMatrixPostProcessingTime(long t) {
+ outMatrixPostProcessingTime.add(t);
+ }
+
+ public static long getEncodeBuildTime() {
+ return binningBuildTime.longValue() + imputeBuildTime.longValue() +
+ recodeBuildTime.longValue();
+ }
+
+ public static long getEncodeApplyTime() {
+ return dummyCodeApplyTime.longValue() + binningApplyTime.longValue() +
+ featureHashingApplyTime.longValue() + passThroughApplyTime.longValue() +
+ recodeApplyTime.longValue() + omitApplyTime.longValue() +
+ imputeApplyTime.longValue();
+ }
+
+ public static void reset() {
+ encoderCount.reset();
+ // buildTime.reset();
+ recodeBuildTime.reset();
+ binningBuildTime.reset();
+ imputeBuildTime.reset();
+ // applyTime.reset();
+ recodeApplyTime.reset();
+ dummyCodeApplyTime.reset();
+ passThroughApplyTime.reset();
+ featureHashingApplyTime.reset();
+ binningApplyTime.reset();
+ omitApplyTime.reset();
+ imputeApplyTime.reset();
+ outMatrixPreProcessingTime.reset();
+ outMatrixPostProcessingTime.reset();
+ }
+
+ public static String displayStatistics() {
+ if( encoderCount.longValue() > 0) {
+ //TODO: Cleanup and condense
+ StringBuilder sb = new StringBuilder();
+ sb.append("TransformEncode num. encoders:\t").append(encoderCount.longValue()).append("\n");
+ sb.append("TransformEncode build time:\t").append(String.format("%.3f",
+ getEncodeBuildTime()*1e-9)).append(" sec.\n");
+ if(recodeBuildTime.longValue() > 0)
+ sb.append("\tRecode build time:\t").append(String.format("%.3f",
+ recodeBuildTime.longValue()*1e-9)).append(" sec.\n");
+ if(binningBuildTime.longValue() > 0)
+ sb.append("\tBinning build time:\t").append(String.format("%.3f",
+ binningBuildTime.longValue()*1e-9)).append(" sec.\n");
+ if(imputeBuildTime.longValue() > 0)
+ sb.append("\tImpute build time:\t").append(String.format("%.3f",
+ imputeBuildTime.longValue()*1e-9)).append(" sec.\n");
+
+ sb.append("TransformEncode apply time:\t").append(String.format("%.3f",
+ getEncodeApplyTime()*1e-9)).append(" sec.\n");
+ if(recodeApplyTime.longValue() > 0)
+ sb.append("\tRecode apply time:\t").append(String.format("%.3f",
+ recodeApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(binningApplyTime.longValue() > 0)
+ sb.append("\tBinning apply time:\t").append(String.format("%.3f",
+ binningApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(dummyCodeApplyTime.longValue() > 0)
+ sb.append("\tDummyCode apply time:\t").append(String.format("%.3f",
+ dummyCodeApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(featureHashingApplyTime.longValue() > 0)
+ sb.append("\tHashing apply time:\t").append(String.format("%.3f",
+ featureHashingApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(passThroughApplyTime.longValue() > 0)
+ sb.append("\tPassThrough apply time:\t").append(String.format("%.3f",
+ passThroughApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(omitApplyTime.longValue() > 0)
+ sb.append("\tOmit apply time:\t").append(String.format("%.3f",
+ omitApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(imputeApplyTime.longValue() > 0)
+ sb.append("\tImpute apply time:\t").append(String.format("%.3f",
+ imputeApplyTime.longValue()*1e-9)).append(" sec.\n");
+
+ sb.append("TransformEncode PreProc. time:\t").append(String.format("%.3f",
+ outMatrixPreProcessingTime.longValue()*1e-9)).append(" sec.\n");
+ sb.append("TransformEncode PostProc. time:\t").append(String.format("%.3f",
+ outMatrixPostProcessingTime.longValue()*1e-9)).append(" sec.\n");
+ return sb.toString();
+ }
+ return "";
+ }
+}
diff --git a/src/test/config/SystemDS-MultiTenant-config.xml b/src/test/config/SystemDS-MultiTenant-config.xml
index 4e2ecdb..ad2bcf0 100644
--- a/src/test/config/SystemDS-MultiTenant-config.xml
+++ b/src/test/config/SystemDS-MultiTenant-config.xml
@@ -22,4 +22,5 @@
<sysds.federated.initialization.timeout>30</sysds.federated.initialization.timeout>
<!-- The timeout of each instruction sent to federated workers -->
<sysds.federated.timeout>128</sysds.federated.timeout>
+ <sysds.local.spark>true</sysds.local.spark>
</root>
diff --git a/src/test/java/org/apache/sysds/test/functions/async/AsyncBroadcastTest.java b/src/test/java/org/apache/sysds/test/functions/async/AsyncBroadcastTest.java
index 088773e..0884753 100644
--- a/src/test/java/org/apache/sysds/test/functions/async/AsyncBroadcastTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/async/AsyncBroadcastTest.java
@@ -32,6 +32,7 @@ import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.test.TestConfiguration;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.SparkStatistics;
import org.junit.Assert;
import org.junit.Test;
@@ -100,7 +101,7 @@ public class AsyncBroadcastTest extends AutomatedTestBase {
long expected_successBC = 1;
long numBC = Statistics.getCPHeavyHitterCount("broadcast");
Assert.assertTrue("Violated Broadcast instruction count: "+numBC, numBC == expected_numBC);
- long successBC = Statistics.getAsyncBroadcastCount();
+ long successBC = SparkStatistics.getAsyncBroadcastCount();
Assert.assertTrue("Violated successful Broadcast count: "+successBC, successBC == expected_successBC);
} finally {
OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = old_simplification;
diff --git a/src/test/java/org/apache/sysds/test/functions/async/PrefetchRDDTest.java b/src/test/java/org/apache/sysds/test/functions/async/PrefetchRDDTest.java
index 7da3f52..5a88472 100644
--- a/src/test/java/org/apache/sysds/test/functions/async/PrefetchRDDTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/async/PrefetchRDDTest.java
@@ -32,6 +32,7 @@ import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.test.TestConfiguration;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.SparkStatistics;
import org.junit.Assert;
import org.junit.Test;
@@ -109,7 +110,7 @@ public class PrefetchRDDTest extends AutomatedTestBase {
long expected_successPF = !testname.equalsIgnoreCase(TEST_NAME+"3") ? 1 : 0;
long numPF = Statistics.getCPHeavyHitterCount("prefetch");
Assert.assertTrue("Violated Prefetch instruction count: "+numPF, numPF == expected_numPF);
- long successPF = Statistics.getAsyncPrefetchCount();
+ long successPF = SparkStatistics.getAsyncPrefetchCount();
Assert.assertTrue("Violated successful Prefetch count: "+successPF, successPF == expected_successPF);
} finally {
OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = old_simplification;
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/LocalInstruction.java b/src/test/java/org/apache/sysds/test/functions/compress/LocalInstruction.java
index 5a2de7c..c7f98f8 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/LocalInstruction.java
+++ b/src/test/java/org/apache/sysds/test/functions/compress/LocalInstruction.java
@@ -29,6 +29,7 @@ import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.SparkStatistics;
import org.junit.Assert;
public abstract class LocalInstruction extends AutomatedTestBase {
@@ -54,6 +55,8 @@ public abstract class LocalInstruction extends AutomatedTestBase {
OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND = true;
OptimizerUtils.ALLOW_SCRIPT_LEVEL_LOCAL_COMMAND = true;
+ setOutputBuffering(true);
+
try {
loadTestConfiguration(getTestConfiguration(testName));
@@ -65,7 +68,7 @@ public abstract class LocalInstruction extends AutomatedTestBase {
String ret = runTest(null).toString();
long actualCompressionCount = Statistics.getCPHeavyHitterCount("sp_compress");
- long actualCollectCount = Statistics.getSparkCollectCount();
+ long actualCollectCount = SparkStatistics.getSparkCollectCount();
Assert.assertEquals(ret + "Compression count is incorrect", compressionCount, actualCompressionCount);
Assert.assertEquals(ret + "Collection count is incorrect", sparkCollectionCount, actualCollectCount);
Statistics.reset();
diff --git a/src/test/java/org/apache/sysds/test/functions/federated/multitenant/FederatedMultiTenantTest.java b/src/test/java/org/apache/sysds/test/functions/federated/multitenant/FederatedMultiTenantTest.java
index eade66c..1b545ff 100644
--- a/src/test/java/org/apache/sysds/test/functions/federated/multitenant/FederatedMultiTenantTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/federated/multitenant/FederatedMultiTenantTest.java
@@ -104,11 +104,10 @@ public class FederatedMultiTenantTest extends AutomatedTestBase {
runMultiTenantSameWorkerTest(OpType.SUM, 4, ExecMode.SPARK);
}
-//FIXME still runs into blocking
-// @Test
-// public void testSumSharedWorkersSP() {
-// runMultiTenantSharedWorkerTest(OpType.SUM, 3, 9, ExecMode.SPARK);
-// }
+ @Test
+ public void testSumSharedWorkersSP() {
+ runMultiTenantSharedWorkerTest(OpType.SUM, 3, 9, ExecMode.SPARK);
+ }
@Test
@Ignore
@@ -342,8 +341,7 @@ public class FederatedMultiTenantTest extends AutomatedTestBase {
argsList.addAll(Arrays.asList(args));
ProcessBuilder processBuilder = new ProcessBuilder(ArrayUtils.addAll(new String[]{
- path, "-cp", classpath, DMLScript.class.getName()}, argsList.toArray(new String[0])))
- .redirectErrorStream(true);
+ path, "-cp", classpath, DMLScript.class.getName()}, argsList.toArray(new String[0])));
Process process = null;
try {
diff --git a/src/test/java/org/apache/sysds/test/functions/jmlc/JMLCParfor2ForCompileTest.java b/src/test/java/org/apache/sysds/test/functions/jmlc/JMLCParfor2ForCompileTest.java
index 69a5e43..1c163ac 100644
--- a/src/test/java/org/apache/sysds/test/functions/jmlc/JMLCParfor2ForCompileTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/jmlc/JMLCParfor2ForCompileTest.java
@@ -27,6 +27,7 @@ import org.apache.sysds.api.jmlc.PreparedScript;
import org.apache.sysds.conf.CompilerConfig.ConfigType;
import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.ParForStatistics;
public class JMLCParfor2ForCompileTest extends AutomatedTestBase
{
@@ -69,6 +70,6 @@ public class JMLCParfor2ForCompileTest extends AutomatedTestBase
}
//check for existing or non-existing parfor
- Assert.assertTrue(Statistics.getParforOptCount()==(par?1:0));
+ Assert.assertTrue(ParForStatistics.getOptCount()==(par?1:0));
}
}
diff --git a/src/test/java/org/apache/sysds/test/functions/recompile/CSVReadInFunctionTest.java b/src/test/java/org/apache/sysds/test/functions/recompile/CSVReadInFunctionTest.java
index 88f1c4a..908512d 100644
--- a/src/test/java/org/apache/sysds/test/functions/recompile/CSVReadInFunctionTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/recompile/CSVReadInFunctionTest.java
@@ -35,6 +35,7 @@ import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.test.TestConfiguration;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.utils.Statistics;
+import org.apache.sysds.utils.stats.SparkStatistics;
public class CSVReadInFunctionTest extends AutomatedTestBase {
@@ -109,7 +110,7 @@ public class CSVReadInFunctionTest extends AutomatedTestBase {
//check no executed spark instructions
Assert.assertEquals(Statistics.getNoOfExecutedSPInst(), 0);
- Assert.assertTrue(!Statistics.createdSparkContext());
+ Assert.assertTrue(!SparkStatistics.createdSparkContext());
}
catch(Exception ex) {
throw new RuntimeException(ex);