[SYSTEMDS-2568] Initial Federated Privacy Support
- Privacy Constraint support for GLM
- Privacy tests for GLM
- Improved exception handling of federated responses
- Log of checked privacy constraints
This is to give the federated master information about which privacy
constraints were violated and to be able to throw the actual exception
on the master side.
Add Initial Implementation of Checked Privacy Constraints Log
This will enable the user to check which privacy constraints were
retrieved during handling of federated instruction. This is an initial
implementation since the checked privacy constraints are added to the
federated response, but this is never retrieved by the federated master.
If the privacy constraint is null for a checked data object, this is
currently not logged. This could easily be changed by moving the put
operation before the privacy constraint null check in the
PrivacyMonitor.
Closes #946
diff --git a/src/main/java/org/apache/sysds/api/DMLOptions.java b/src/main/java/org/apache/sysds/api/DMLOptions.java
index b9972a3..be0e266 100644
--- a/src/main/java/org/apache/sysds/api/DMLOptions.java
+++ b/src/main/java/org/apache/sysds/api/DMLOptions.java
@@ -63,6 +63,7 @@
public ReuseCacheType linReuseType = ReuseCacheType.NONE;
public boolean fedWorker = false;
public int fedWorkerPort = -1;
+ public boolean checkPrivacy = false; // Check which privacy constraints are loaded and checked during federated execution
public final static DMLOptions defaultOptions = new DMLOptions(null);
@@ -226,6 +227,8 @@
}
}
+ dmlOptions.checkPrivacy = line.hasOption("checkPrivacy");
+
return dmlOptions;
}
@@ -273,6 +276,9 @@
.hasOptionalArgs().create("lineage");
Option fedOpt = OptionBuilder.withDescription("starts a federated worker with the given argument as the port.")
.hasOptionalArg().create("w");
+ Option checkPrivacy = OptionBuilder
+ .withDescription("Check which privacy constraints are loaded and checked during federated execution")
+ .create("checkPrivacy");
options.addOption(configOpt);
options.addOption(cleanOpt);
@@ -285,6 +291,7 @@
options.addOption(pythonOpt);
options.addOption(lineageOpt);
options.addOption(fedOpt);
+ options.addOption(checkPrivacy);
// Either a clean(-clean), a file(-f), a script(-s) or help(-help) needs to be specified
OptionGroup fileOrScriptOpt = new OptionGroup()
diff --git a/src/main/java/org/apache/sysds/api/DMLScript.java b/src/main/java/org/apache/sysds/api/DMLScript.java
index 6854ebb..477f0ec 100644
--- a/src/main/java/org/apache/sysds/api/DMLScript.java
+++ b/src/main/java/org/apache/sysds/api/DMLScript.java
@@ -65,6 +65,7 @@
import org.apache.sysds.runtime.io.IOUtilFunctions;
import org.apache.sysds.runtime.lineage.LineageCacheConfig;
import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
+import org.apache.sysds.runtime.privacy.CheckedConstraintsLog;
import org.apache.sysds.runtime.util.LocalFileUtils;
import org.apache.sysds.runtime.util.HDFSTool;
import org.apache.sysds.utils.Explain;
@@ -92,6 +93,7 @@
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 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;
@@ -192,6 +194,7 @@
LINEAGE = dmlOptions.lineage;
LINEAGE_DEDUP = dmlOptions.lineage_dedup;
LINEAGE_REUSE = dmlOptions.linReuseType;
+ CHECK_PRIVACY = dmlOptions.checkPrivacy;
String fnameOptConfig = dmlOptions.configFile;
boolean isFile = dmlOptions.filePath != null;
@@ -464,6 +467,8 @@
Statistics.resetNoOfExecutedJobs();
if( STATISTICS )
Statistics.reset();
+ if ( CHECK_PRIVACY )
+ CheckedConstraintsLog.reset();
}
public static void cleanupHadoopExecution( DMLConfig config )
diff --git a/src/main/java/org/apache/sysds/api/mlcontext/ScriptExecutor.java b/src/main/java/org/apache/sysds/api/mlcontext/ScriptExecutor.java
index bf9c4ee..b9fba88 100644
--- a/src/main/java/org/apache/sysds/api/mlcontext/ScriptExecutor.java
+++ b/src/main/java/org/apache/sysds/api/mlcontext/ScriptExecutor.java
@@ -51,6 +51,7 @@
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContextFactory;
import org.apache.sysds.runtime.lineage.LineageItemUtils;
+import org.apache.sysds.runtime.privacy.CheckedConstraintsLog;
import org.apache.sysds.utils.Explain;
import org.apache.sysds.utils.Statistics;
import org.apache.sysds.utils.Explain.ExplainCounts;
@@ -374,6 +375,8 @@
Statistics.resetNoOfExecutedJobs();
if (statistics)
Statistics.reset();
+ if ( DMLScript.CHECK_PRIVACY )
+ CheckedConstraintsLog.reset();
}
/**
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java
index 74ad0d3..19c33a9 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java
@@ -175,8 +175,6 @@
FederatedRange range = readResponse.getLeft();
FederatedResponse response = readResponse.getRight().get();
// add result
- if(!response.isSuccessful())
- throw new DMLRuntimeException("Federated matrix read failed: " + response.getErrorMessage());
FrameBlock multRes = (FrameBlock) response.getData()[0];
for (int r = 0; r < multRes.getNumRows(); r++) {
for (int c = 0; c < multRes.getNumColumns(); c++) {
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/MatrixObject.java b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/MatrixObject.java
index c5b8822..5848e9e 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/caching/MatrixObject.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/caching/MatrixObject.java
@@ -41,6 +41,7 @@
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
import org.apache.sysds.runtime.meta.MetaData;
import org.apache.sysds.runtime.meta.MetaDataFormat;
+import org.apache.sysds.runtime.privacy.DMLPrivacyException;
import org.apache.sysds.runtime.util.DataConverter;
import org.apache.sysds.runtime.util.HDFSTool;
import org.apache.sysds.runtime.util.IndexRange;
@@ -421,8 +422,6 @@
// add result
int[] beginDimsInt = range.getBeginDimsInt();
int[] endDimsInt = range.getEndDimsInt();
- if( !response.isSuccessful() )
- throw new DMLRuntimeException("Federated matrix read failed: " + response.getErrorMessage());
MatrixBlock multRes = (MatrixBlock) response.getData()[0];
result.copy(beginDimsInt[0], endDimsInt[0] - 1,
beginDimsInt[1], endDimsInt[1] - 1, multRes, false);
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 8e59431..ce07488 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
@@ -24,6 +24,8 @@
import java.util.Arrays;
import java.util.List;
+import org.apache.sysds.api.DMLScript;
+
public class FederatedRequest implements Serializable {
private static final long serialVersionUID = 5946781306963870394L;
@@ -33,20 +35,24 @@
private FedMethod _method;
private List<Object> _data;
+ private boolean checkPrivacy;
public FederatedRequest(FedMethod method, List<Object> data) {
_method = method;
_data = data;
+ setCheckPrivacy();
}
public FederatedRequest(FedMethod method, Object ... datas) {
_method = method;
_data = Arrays.asList(datas);
+ setCheckPrivacy();
}
public FederatedRequest(FedMethod method) {
_method = method;
_data = new ArrayList<>();
+ setCheckPrivacy();
}
public FedMethod getMethod() {
@@ -74,4 +80,16 @@
public FederatedRequest deepClone() {
return new FederatedRequest(_method, new ArrayList<>(_data));
}
+
+ public void setCheckPrivacy(boolean checkPrivacy){
+ this.checkPrivacy = checkPrivacy;
+ }
+
+ public void setCheckPrivacy(){
+ setCheckPrivacy(DMLScript.CHECK_PRIVACY);
+ }
+
+ public boolean checkPrivacy(){
+ return checkPrivacy;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedResponse.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedResponse.java
index 6032984..c187051 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedResponse.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedResponse.java
@@ -20,6 +20,14 @@
package org.apache.sysds.runtime.controlprogram.federated;
import java.io.Serializable;
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.privacy.CheckedConstraintsLog;
+import org.apache.sysds.runtime.privacy.PrivacyConstraint.PrivacyLevel;
public class FederatedResponse implements Serializable {
private static final long serialVersionUID = 3142180026498695091L;
@@ -32,6 +40,7 @@
private FederatedResponse.Type _status;
private Object[] _data;
+ private Map<PrivacyLevel,LongAdder> checkedConstraints;
public FederatedResponse(FederatedResponse.Type status) {
this(status, null);
@@ -56,10 +65,45 @@
}
public String getErrorMessage() {
- return (String) _data[0];
+ return ExceptionUtils.getFullStackTrace( (Exception) _data[0] );
}
- public Object[] getData() {
+ public Object[] getData() throws Exception {
+ updateCheckedConstraintsLog();
+ if ( !isSuccessful() )
+ throwExceptionFromResponse();
return _data;
}
+
+ /**
+ * Checks the data object array for exceptions that occurred in the federated worker
+ * during handling of request.
+ * @throws Exception the exception retrieved from the data object array
+ * or DMLRuntimeException if no exception is provided by the federated worker.
+ */
+ public void throwExceptionFromResponse() throws Exception {
+ for ( Object potentialException : _data){
+ if (potentialException != null && (potentialException instanceof Exception) ){
+ throw (Exception) potentialException;
+ }
+ }
+ throw new DMLRuntimeException("Unknown runtime exception in handling of federated request by federated worker.");
+ }
+
+ /**
+ * Set checked privacy constraints in response if the provided map is not empty.
+ * If the map is empty, it means that no privacy constraints were found.
+ * @param checkedConstraints map of checked constraints from the PrivacyMonitor
+ */
+ public void setCheckedConstraints(Map<PrivacyLevel,LongAdder> checkedConstraints){
+ if ( checkedConstraints != null && !checkedConstraints.isEmpty() ){
+ this.checkedConstraints = new EnumMap<PrivacyLevel, LongAdder>(PrivacyLevel.class);
+ this.checkedConstraints.putAll(checkedConstraints);
+ }
+ }
+
+ public void updateCheckedConstraintsLog(){
+ if ( checkedConstraints != null && !checkedConstraints.isEmpty() )
+ CheckedConstraintsLog.addCheckedConstraints(checkedConstraints);
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java
index 6fe814a..27e20e2 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java
@@ -23,7 +23,6 @@
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
-import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
@@ -80,15 +79,23 @@
throw new DMLRuntimeException("FederatedWorkerHandler: Received object no instance of `FederatedRequest`.");
FederatedRequest.FedMethod method = request.getMethod();
log.debug("Received command: " + method.name());
+ PrivacyMonitor.setCheckPrivacy(request.checkPrivacy());
+ PrivacyMonitor.clearCheckedConstraints();
synchronized (_seq) {
FederatedResponse response = constructResponse(request);
+ conditionalAddCheckedConstraints(request, response);
if (!response.isSuccessful())
log.error("Method " + method + " failed: " + response.getErrorMessage());
ctx.writeAndFlush(response).addListener(new CloseListener());
}
}
+ private void conditionalAddCheckedConstraints(FederatedRequest request, FederatedResponse response){
+ if ( request.checkPrivacy() )
+ response.setCheckedConstraints(PrivacyMonitor.getCheckedConstraints());
+ }
+
private FederatedResponse constructResponse(FederatedRequest request) {
FederatedRequest.FedMethod method = request.getMethod();
try {
@@ -111,7 +118,7 @@
}
}
catch (Exception exception) {
- return new FederatedResponse(FederatedResponse.Type.ERROR, ExceptionUtils.getFullStackTrace(exception));
+ return new FederatedResponse(FederatedResponse.Type.ERROR, exception);
}
}
@@ -318,6 +325,7 @@
public void operationComplete(ChannelFuture channelFuture) throws InterruptedException, DMLRuntimeException {
if (!channelFuture.isSuccess())
throw new DMLRuntimeException("Federated Worker Write failed");
+ PrivacyMonitor.clearCheckedConstraints();
channelFuture.channel().close().sync();
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/LibFederatedAgg.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/LibFederatedAgg.java
index 1f40221..feb5a68 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/LibFederatedAgg.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/LibFederatedAgg.java
@@ -66,8 +66,6 @@
FederatedRange range = idResponsePair.getLeft();
FederatedResponse federatedResponse = idResponsePair.getRight().get();
int[] beginDims = range.getBeginDimsInt();
- if (!federatedResponse.isSuccessful())
- throw new DMLRuntimeException("Federated aggregation failed: " + federatedResponse.getErrorMessage());
MatrixBlock mb = (MatrixBlock) federatedResponse.getData()[0];
// TODO performance optimizations
MatrixValue.CellIndex cellIndex = new MatrixValue.CellIndex(0, 0);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/FunctionCallCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/FunctionCallCPInstruction.java
index f00f42d..695b07a 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/FunctionCallCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/FunctionCallCPInstruction.java
@@ -270,6 +270,10 @@
return sb.substring( 0, sb.length()-Lop.OPERAND_DELIMITOR.length() );
}
+
+ public CPOperand[] getInputs(){
+ return _boundInputs;
+ }
private boolean reuseFunctionOutputs(LineageItem[] liInputs, FunctionProgramBlock fpb, ExecutionContext ec) {
//prepare lineage cache probing
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateBinaryFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateBinaryFEDInstruction.java
index caed372..2c064fc 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateBinaryFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateBinaryFEDInstruction.java
@@ -210,20 +210,24 @@
private static void combinePartialMVResults(FederatedRange range,
FederatedResponse federatedResponse, MatrixBlock resultBlock, boolean matrixVectorOp)
{
- int[] beginDims = range.getBeginDimsInt();
- MatrixBlock mb = (MatrixBlock) federatedResponse.getData()[0];
- // TODO performance optimizations
- // TODO Improve Vector Matrix multiplication accuracy: An idea would be to make use of kahan plus here,
- // this should improve accuracy a bit, although we still lose out on the small error lost on the worker
- // without having to return twice the amount of data (value + sum error)
- // Add worker response to resultBlock
- for (int r = 0; r < mb.getNumRows(); r++)
- for (int c = 0; c < mb.getNumColumns(); c++) {
- int resultRow = r + (!matrixVectorOp ? 0 : beginDims[0]);
- int resultColumn = c + (!matrixVectorOp ? beginDims[1] : 0);
- resultBlock.quickSetValue(resultRow, resultColumn,
- resultBlock.quickGetValue(resultRow, resultColumn) + mb.quickGetValue(r, c));
- }
+ try {
+ int[] beginDims = range.getBeginDimsInt();
+ MatrixBlock mb = (MatrixBlock) federatedResponse.getData()[0];
+ // TODO performance optimizations
+ // TODO Improve Vector Matrix multiplication accuracy: An idea would be to make use of kahan plus here,
+ // this should improve accuracy a bit, although we still lose out on the small error lost on the worker
+ // without having to return twice the amount of data (value + sum error)
+ // Add worker response to resultBlock
+ for (int r = 0; r < mb.getNumRows(); r++)
+ for (int c = 0; c < mb.getNumColumns(); c++) {
+ int resultRow = r + (!matrixVectorOp ? 0 : beginDims[0]);
+ int resultColumn = c + (!matrixVectorOp ? beginDims[1] : 0);
+ resultBlock.quickSetValue(resultRow, resultColumn,
+ resultBlock.quickGetValue(resultRow, resultColumn) + mb.quickGetValue(r, c));
+ }
+ } catch (Exception e){
+ throw new DMLRuntimeException("Combine partial results from federated matrix failed.", e);
+ }
}
private static Future<FederatedResponse> executeMVMultiply(FederatedRange range,
@@ -323,12 +327,12 @@
// TODO experiment if sending multiple requests at the same time to the same worker increases
// performance (remove get and do multithreaded?)
FederatedResponse response = executeMVMultiply(_range, _data, vec, _distributeCols).get();
- if(response.isSuccessful()) {
+ try{
result.copy(r, r, 0, endDims[1] - beginDims[1] - 1, (MatrixBlock) response.getData()[0], true);
- }
- else
+ } catch (Exception e) {
throw new DMLRuntimeException(
- "Federated Matrix-Matrix Multiplication failed: " + response.getErrorMessage());
+ "Federated Matrix-Matrix Multiplication failed: ", e);
+ }
}
_result.setRight(result);
}
@@ -366,12 +370,13 @@
// TODO experiment if sending multiple requests at the same time to the same worker increases
// performance
FederatedResponse response = executeMVMultiply(_range, _data, vec, _distributeCols).get();
- if(response.isSuccessful()) {
+ try {
result.copy(0, endDims[0] - beginDims[0] - 1, c, c, (MatrixBlock) response.getData()[0], true);
- }
- else
+ } catch (Exception e){
throw new DMLRuntimeException(
- "Federated Matrix-Matrix Multiplication failed: " + response.getErrorMessage());
+ "Federated Matrix-Matrix Multiplication failed: ", e);
+ }
+
}
_result.setRight(result);
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixScalarFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixScalarFEDInstruction.java
index 74fd1ab..3c9db11 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixScalarFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixScalarFEDInstruction.java
@@ -76,8 +76,6 @@
FederatedRange range = idResponsePair.getLeft();
//wait for fed workers finishing their work
FederatedResponse federatedResponse = idResponsePair.getRight().get();
- if (!federatedResponse.isSuccessful())
- throw new DMLRuntimeException("Federated binary operation failed: " + federatedResponse.getErrorMessage());
MatrixBlock shard = (MatrixBlock) federatedResponse.getData()[0];
ret.copy(range.getBeginDimsInt()[0], range.getEndDimsInt()[0]-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 5e0eb9c..e4b7dac 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
@@ -223,10 +223,7 @@
try {
for (Pair<FederatedData, Future<FederatedResponse>> idResponse : idResponses) {
FederatedResponse response = idResponse.getRight().get();
- if (response.isSuccessful())
- idResponse.getLeft().setVarID((Long) response.getData()[0]);
- else
- throw new DMLRuntimeException(response.getErrorMessage());
+ idResponse.getLeft().setVarID((Long) response.getData()[0]);
}
}
catch (Exception e) {
@@ -279,11 +276,12 @@
private static void handleFedFrameResponse(Types.ValueType[] schema, FederatedData federatedData,
FederatedResponse response, int startColumn) {
- if(response.isSuccessful()) {
+ try {
// Index 0 is the varID, Index 1 is the schema of the frame
- federatedData.setVarID((Long) response.getData()[0]);
+ Object[] data = response.getData();
+ federatedData.setVarID((Long) data[0]);
// copy the
- Types.ValueType[] range_schema = (Types.ValueType[]) response.getData()[1];
+ Types.ValueType[] range_schema = (Types.ValueType[]) data[1];
for(int i = 0; i < range_schema.length; i++) {
Types.ValueType vType = range_schema[i];
int schema_index = startColumn + i;
@@ -292,8 +290,8 @@
else
schema[schema_index] = vType;
}
+ } catch (Exception e){
+ throw new DMLRuntimeException("Exception in frame response from federated worker.", e);
}
- else
- throw new DMLRuntimeException(response.getErrorMessage());
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/privacy/CheckedConstraintsLog.java b/src/main/java/org/apache/sysds/runtime/privacy/CheckedConstraintsLog.java
new file mode 100644
index 0000000..e6bc7c0
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/privacy/CheckedConstraintsLog.java
@@ -0,0 +1,67 @@
+/*
+ * 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.runtime.privacy;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.BiFunction;
+
+import org.apache.sysds.runtime.privacy.PrivacyConstraint.PrivacyLevel;
+
+public class CheckedConstraintsLog {
+ private static Map<PrivacyLevel,LongAdder> checkedConstraintsTotal = new EnumMap<PrivacyLevel,LongAdder>(PrivacyLevel.class);
+ private static BiFunction<LongAdder, LongAdder, LongAdder> mergeLongAdders = (v1, v2) -> {
+ v1.add(v2.longValue() );
+ return v1;
+ };
+
+ /**
+ * Adds checkedConstraints to the checked constraints total.
+ * @param checkedConstraints constraints checked by federated worker
+ */
+ public static void addCheckedConstraints(Map<PrivacyLevel,LongAdder> checkedConstraints){
+ if ( checkedConstraints != null){
+ checkedConstraints.forEach(
+ (key,value) -> checkedConstraintsTotal.merge( key, value, mergeLongAdders) );
+ }
+ }
+
+ /**
+ * Remove all elements from checked constraints log.
+ */
+ public static void reset(){
+ checkedConstraintsTotal.clear();
+ }
+
+ public static Map<PrivacyLevel,LongAdder> getCheckedConstraints(){
+ return checkedConstraintsTotal;
+ }
+
+ /**
+ * Get string representing all contents of the checked constraints log.
+ * @return string representation of checked constraints log.
+ */
+ public static String display(){
+ StringBuilder sb = new StringBuilder();
+ checkedConstraintsTotal.forEach((k,v)->sb.append("\t" + k + ": " + v + "\n"));
+ return sb.toString();
+ }
+}
\ No newline at end of file
diff --git a/src/main/java/org/apache/sysds/runtime/privacy/PrivacyMonitor.java b/src/main/java/org/apache/sysds/runtime/privacy/PrivacyMonitor.java
index ee88bf4..3978b6d 100644
--- a/src/main/java/org/apache/sysds/runtime/privacy/PrivacyMonitor.java
+++ b/src/main/java/org/apache/sysds/runtime/privacy/PrivacyMonitor.java
@@ -19,6 +19,10 @@
package org.apache.sysds.runtime.privacy;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.concurrent.atomic.LongAdder;
+
import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
@@ -27,10 +31,39 @@
import org.apache.sysds.runtime.privacy.PrivacyConstraint.PrivacyLevel;
public class PrivacyMonitor
-{
- //TODO maybe maintain a log of checked constaints for transfers
- // in order to provide 'privacy explanations' similar to our stats
-
+{
+ private static EnumMap<PrivacyLevel,LongAdder> checkedConstraints;
+
+ static {
+ checkedConstraints = new EnumMap<PrivacyLevel,LongAdder>(PrivacyLevel.class);
+ for ( PrivacyLevel level : PrivacyLevel.values() ){
+ checkedConstraints.put(level, new LongAdder());
+ }
+ }
+
+ private static boolean checkPrivacy = false;
+
+ public static EnumMap<PrivacyLevel,LongAdder> getCheckedConstraints(){
+ return checkedConstraints;
+ }
+
+ private static void incrementCheckedConstraints(PrivacyLevel privacyLevel){
+ if ( checkPrivacy ){
+ if ( privacyLevel == null )
+ throw new NullPointerException("Cannot increment checked constraints log: Privacy level is null.");
+ checkedConstraints.get(privacyLevel).increment();
+ }
+
+ }
+
+ public static void clearCheckedConstraints(){
+ checkedConstraints.replaceAll((k,v)->new LongAdder());
+ }
+
+ public static void setCheckPrivacy(boolean checkPrivacyParam){
+ checkPrivacy = checkPrivacyParam;
+ }
+
/**
* Throws DMLPrivacyException if data object is CacheableData and privacy constraint is set to private or private aggregation.
* @param dataObject input data object
@@ -41,12 +74,13 @@
PrivacyConstraint privacyConstraint = ((CacheableData<?>)dataObject).getPrivacyConstraint();
if (privacyConstraint != null){
PrivacyLevel privacyLevel = privacyConstraint.getPrivacyLevel();
+ incrementCheckedConstraints(privacyLevel);
switch(privacyLevel){
case None:
((CacheableData<?>)dataObject).setPrivacyConstraints(null);
break;
case Private:
- case PrivateAggregation:
+ case PrivateAggregation:
throw new DMLPrivacyException("Cannot share variable, since the privacy constraint of the requested variable is set to " + privacyLevel.name());
default:
throw new DMLPrivacyException("Privacy level " + privacyLevel.name() + " of variable not recognized");
@@ -65,6 +99,7 @@
PrivacyConstraint privacyConstraint = matrixObject.getPrivacyConstraint();
if (privacyConstraint != null){
PrivacyLevel privacyLevel = privacyConstraint.getPrivacyLevel();
+ incrementCheckedConstraints(privacyLevel);
switch(privacyLevel){
case None:
matrixObject.setPrivacyConstraints(null);
@@ -88,8 +123,11 @@
Data data = ec.getVariable(input);
if ( data != null && (data instanceof CacheableData<?>)){
PrivacyConstraint privacyConstraintIn = ((CacheableData<?>) data).getPrivacyConstraint();
- if ( privacyConstraintIn != null && (privacyConstraintIn.getPrivacyLevel() == PrivacyLevel.Private) ){
- throw new DMLPrivacyException("Privacy constraint cannot be propagated to scalar for input " + input.getName());
+ if ( privacyConstraintIn != null ) {
+ incrementCheckedConstraints(privacyConstraintIn.getPrivacyLevel());
+ if ( privacyConstraintIn.getPrivacyLevel() == PrivacyLevel.Private ){
+ throw new DMLPrivacyException("Privacy constraint cannot be propagated to scalar for input " + input.getName());
+ }
}
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/privacy/PrivacyPropagator.java b/src/main/java/org/apache/sysds/runtime/privacy/PrivacyPropagator.java
index 323330a..298b1a7 100644
--- a/src/main/java/org/apache/sysds/runtime/privacy/PrivacyPropagator.java
+++ b/src/main/java/org/apache/sysds/runtime/privacy/PrivacyPropagator.java
@@ -19,8 +19,8 @@
package org.apache.sysds.runtime.privacy;
-import java.util.function.Function;
+import org.apache.sysds.common.Types.DataType;
import org.apache.sysds.parser.DataExpression;
import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
@@ -31,6 +31,7 @@
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.ComputationCPInstruction;
import org.apache.sysds.runtime.instructions.cp.Data;
+import org.apache.sysds.runtime.instructions.cp.FunctionCallCPInstruction;
import org.apache.sysds.runtime.instructions.cp.QuaternaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.UnaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.VariableCPInstruction;
@@ -78,14 +79,6 @@
return null;
}
- public static PrivacyConstraint mergeTernary(PrivacyConstraint[] privacyConstraints){
- return mergeBinary(mergeBinary(privacyConstraints[0], privacyConstraints[1]), privacyConstraints[2]);
- }
-
- public static PrivacyConstraint mergeQuaternary(PrivacyConstraint[] privacyConstraints){
- return mergeBinary(mergeTernary(privacyConstraints), privacyConstraints[3]);
- }
-
public static PrivacyConstraint mergeNary(PrivacyConstraint[] privacyConstraints){
PrivacyConstraint mergedPrivacyConstraint = privacyConstraints[0];
for ( int i = 1; i < privacyConstraints.length; i++ ){
@@ -130,6 +123,8 @@
case BuiltinNary:
case Builtin:
return preprocessBuiltinNary((BuiltinNaryCPInstruction) inst, ec);
+ case External:
+ return preprocessExternal((FunctionCallCPInstruction) inst, ec);
case Ctable:
case MultiReturnParameterizedBuiltin:
case MultiReturnBuiltin:
@@ -150,39 +145,56 @@
return inst;
}
- public static Instruction preprocessBuiltinNary(BuiltinNaryCPInstruction inst, ExecutionContext ec){
- if (inst.getInputs() == null) return inst;
- PrivacyConstraint[] privacyConstraints = getInputPrivacyConstraints(ec, inst.getInputs());
- PrivacyConstraint mergedPrivacyConstraint = mergeNary(privacyConstraints);
- inst.setPrivacyConstraint(mergedPrivacyConstraint);
- setOutputPrivacyConstraint(ec, mergedPrivacyConstraint, inst.getOutput());
+
+ public static Instruction preprocessExternal(FunctionCallCPInstruction inst, ExecutionContext ec){
+ return mergePrivacyConstraintsFromInput(
+ inst,
+ ec,
+ inst.getInputs(),
+ inst.getBoundOutputParamNames().toArray(new String[0])
+ );
+ }
+
+ private static Instruction mergePrivacyConstraintsFromInput(Instruction inst, ExecutionContext ec, CPOperand[] inputs, String[] outputNames){
+ if ( inputs != null && inputs.length > 0 ){
+ PrivacyConstraint[] privacyConstraints = getInputPrivacyConstraints(ec, inputs);
+ if ( privacyConstraints != null ){
+ PrivacyConstraint mergedPrivacyConstraint = mergeNary(privacyConstraints);
+ inst.setPrivacyConstraint(mergedPrivacyConstraint);
+ if ( outputNames != null ){
+ for (String outputName : outputNames)
+ setOutputPrivacyConstraint(ec, mergedPrivacyConstraint, outputName);
+ }
+ }
+ }
return inst;
}
+ private static Instruction mergePrivacyConstraintsFromInput(Instruction inst, ExecutionContext ec, CPOperand[] inputs, CPOperand output){
+ String outputName = (output != null) ? output.getName() : null;
+ return mergePrivacyConstraintsFromInput(inst, ec, inputs, new String[]{outputName});
+ }
+
+ public static Instruction preprocessBuiltinNary(BuiltinNaryCPInstruction inst, ExecutionContext ec){
+ return mergePrivacyConstraintsFromInput(inst, ec, inst.getInputs(), inst.getOutput() );
+ }
+
public static Instruction preprocessQuaternary(QuaternaryCPInstruction inst, ExecutionContext ec){
- PrivacyConstraint[] privacyConstraints = getInputPrivacyConstraints(ec,
- new CPOperand[] {inst.input1,inst.input2,inst.input3,inst.getInput4()});
- PrivacyConstraint mergedPrivacyConstraint = mergeQuaternary(privacyConstraints);
- inst.setPrivacyConstraint(mergedPrivacyConstraint);
- setOutputPrivacyConstraint(ec, mergedPrivacyConstraint, inst.output);
- return inst;
+ return mergePrivacyConstraintsFromInput(
+ inst,
+ ec,
+ new CPOperand[] {inst.input1,inst.input2,inst.input3,inst.getInput4()},
+ inst.output
+ );
}
public static Instruction preprocessTernaryCPInstruction(ComputationCPInstruction inst, ExecutionContext ec){
- PrivacyConstraint[] privacyConstraints = getInputPrivacyConstraints(ec, new CPOperand[]{inst.input1, inst.input2, inst.input3});
- PrivacyConstraint mergedPrivacyConstraint = mergeTernary(privacyConstraints);
- inst.setPrivacyConstraint(mergedPrivacyConstraint);
- setOutputPrivacyConstraint(ec, mergedPrivacyConstraint, inst.output);
- return inst;
- }
-
- public static Instruction preprocessNaryInstruction(CPInstruction inst, ExecutionContext ec, CPOperand[] inputs, CPOperand output, Function<PrivacyConstraint[], PrivacyConstraint> mergeFunction){
- PrivacyConstraint[] privacyConstraints = getInputPrivacyConstraints(ec, inputs);
- PrivacyConstraint mergedPrivacyConstraint = mergeFunction.apply(privacyConstraints);
- inst.setPrivacyConstraint(mergedPrivacyConstraint);
- setOutputPrivacyConstraint(ec, mergedPrivacyConstraint, output);
- return inst;
-
+ return mergePrivacyConstraintsFromInput(
+ inst,
+ ec,
+ new CPOperand[]{inst.input1, inst.input2, inst.input3},
+ inst.output
+ );
}
public static Instruction preprocessBinaryCPInstruction(BinaryCPInstruction inst, ExecutionContext ec){
@@ -238,7 +250,7 @@
}
private static void throwExceptionIfPrivacyActivated(Instruction inst, ExecutionContext ec){
- if ( inst.getPrivacyConstraint() != null && inst.getPrivacyConstraint().getPrivacyLevel().equals(PrivacyLevel.Private) ) {
+ if ( inst.getPrivacyConstraint() != null && inst.getPrivacyConstraint().getPrivacyLevel() == PrivacyLevel.Private ) {
throw new DMLPrivacyException("Instruction " + inst + " has privacy constraints activated, but the constraints are not propagated during preprocessing of instruction.");
}
}
@@ -261,10 +273,12 @@
* @return instruction with or without privacy constraints
*/
private static Instruction propagateAllInputPrivacy(VariableCPInstruction inst, ExecutionContext ec){
- //TODO: Propagate the most restricting constraints instead of just the latest activated constraint
- for ( CPOperand input : inst.getInputs() )
- inst = (VariableCPInstruction) propagateInputPrivacy(inst, ec, input, inst.getOutput());
- return inst;
+ return mergePrivacyConstraintsFromInput(
+ inst,
+ ec,
+ inst.getInputs().toArray(new CPOperand[0]),
+ inst.getOutput()
+ );
}
/**
@@ -322,22 +336,33 @@
private static PrivacyConstraint[] getInputPrivacyConstraints(ExecutionContext ec, CPOperand[] inputs){
- PrivacyConstraint[] privacyConstraints = new PrivacyConstraint[inputs.length];
- for ( int i = 0; i < inputs.length; i++ ){
- privacyConstraints[i] = getInputPrivacyConstraint(ec, inputs[i]);
+ if ( inputs != null && inputs.length > 0){
+ boolean privacyFound = false;
+ PrivacyConstraint[] privacyConstraints = new PrivacyConstraint[inputs.length];
+ for ( int i = 0; i < inputs.length; i++ ){
+ privacyConstraints[i] = getInputPrivacyConstraint(ec, inputs[i]);
+ if ( privacyConstraints[i] != null )
+ privacyFound = true;
+ }
+ if ( privacyFound )
+ return privacyConstraints;
}
- return privacyConstraints;
-
+ return null;
}
private static void setOutputPrivacyConstraint(ExecutionContext ec, PrivacyConstraint privacyConstraint, CPOperand output){
- Data dd = ec.getVariable(output.getName());
- if ( dd != null ){
+ setOutputPrivacyConstraint(ec, privacyConstraint, output.getName());
+ }
+
+ private static void setOutputPrivacyConstraint(ExecutionContext ec, PrivacyConstraint privacyConstraint, String outputName){
+ Data dd = ec.getVariable(outputName);
+ if ( dd != null && privacyConstraint != null ){
if ( dd instanceof CacheableData ){
((CacheableData<?>) dd).setPrivacyConstraints(privacyConstraint);
- ec.setVariable(output.getName(), dd);
- }
- else throw new DMLPrivacyException("Privacy constraint of " + output + " cannot be set since it is not an instance of CacheableData");
+ ec.setVariable(outputName, dd);
+ } else if ( privacyConstraint.privacyLevel == PrivacyLevel.Private || !(dd.getDataType() == DataType.SCALAR) )
+ throw new DMLPrivacyException("Privacy constraint of " + outputName + " cannot be set since it is not an instance of CacheableData and it is not a scalar with privacy level " + PrivacyLevel.PrivateAggregation.name() );
+ // if privacy level is PrivateAggregation and data is scalar, the call should pass without propagating any constraints
}
}
}
\ No newline at end of file
diff --git a/src/main/java/org/apache/sysds/utils/Statistics.java b/src/main/java/org/apache/sysds/utils/Statistics.java
index 6ad3e7a..7130004 100644
--- a/src/main/java/org/apache/sysds/utils/Statistics.java
+++ b/src/main/java/org/apache/sysds/utils/Statistics.java
@@ -44,6 +44,7 @@
import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
import org.apache.sysds.runtime.lineage.LineageCacheStatistics;
import org.apache.sysds.runtime.matrix.data.LibMatrixDNN;
+import org.apache.sysds.runtime.privacy.CheckedConstraintsLog;
/**
* This class captures all statistics.
@@ -997,6 +998,9 @@
sb.append("Heavy hitter instructions:\n" + getHeavyHitters(maxHeavyHitters));
}
+ if (DMLScript.CHECK_PRIVACY)
+ sb.append("Checked Privacy Constraints:\n" + CheckedConstraintsLog.display());
+
return sb.toString();
}
}
diff --git a/src/test/java/org/apache/sysds/test/AutomatedTestBase.java b/src/test/java/org/apache/sysds/test/AutomatedTestBase.java
index 0dd92a2..a66ee1e 100644
--- a/src/test/java/org/apache/sysds/test/AutomatedTestBase.java
+++ b/src/test/java/org/apache/sysds/test/AutomatedTestBase.java
@@ -47,7 +47,9 @@
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.data.MatrixValue.CellIndex;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
+import org.apache.sysds.runtime.privacy.CheckedConstraintsLog;
import org.apache.sysds.runtime.privacy.PrivacyConstraint;
+import org.apache.sysds.runtime.privacy.PrivacyConstraint.PrivacyLevel;
import org.apache.sysds.runtime.util.DataConverter;
import org.apache.sysds.runtime.util.HDFSTool;
import org.apache.sysds.utils.ParameterBuilder;
@@ -61,6 +63,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -1789,6 +1792,21 @@
return (count >= minCount);
}
+ protected boolean checkedPrivacyConstraintsContains(PrivacyLevel... levels){
+ for ( PrivacyLevel level : levels)
+ if (!(CheckedConstraintsLog.getCheckedConstraints().containsKey(level)))
+ return false;
+ return true;
+ }
+
+ protected boolean checkedPrivacyConstraintsAbove(Map<PrivacyLevel,Long> levelCounts){
+ for ( Map.Entry<PrivacyLevel,Long> levelCount : levelCounts.entrySet()){
+ if (!(CheckedConstraintsLog.getCheckedConstraints().get(levelCount.getKey()).longValue() >= levelCount.getValue()))
+ return false;
+ }
+ return true;
+ }
+
/**
* Create a SystemDS-preferred Spark Session.
*
diff --git a/src/test/java/org/apache/sysds/test/functions/privacy/CheckedConstraintsLogTest.java b/src/test/java/org/apache/sysds/test/functions/privacy/CheckedConstraintsLogTest.java
new file mode 100644
index 0000000..d187436
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/functions/privacy/CheckedConstraintsLogTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.test.functions.privacy;
+
+import java.util.EnumMap;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.apache.sysds.runtime.privacy.CheckedConstraintsLog;
+import org.apache.sysds.runtime.privacy.PrivacyConstraint.PrivacyLevel;
+import org.apache.sysds.test.AutomatedTestBase;
+import org.junit.Test;
+
+@net.jcip.annotations.NotThreadSafe
+public class CheckedConstraintsLogTest extends AutomatedTestBase {
+
+ @Override
+ public void setUp() {
+ CheckedConstraintsLog.getCheckedConstraints().clear();
+ }
+
+ @Test
+ public void addCheckedConstraintsNull(){
+ CheckedConstraintsLog.addCheckedConstraints(null);
+ assert(CheckedConstraintsLog.getCheckedConstraints() != null && CheckedConstraintsLog.getCheckedConstraints().isEmpty());
+ }
+
+ @Test
+ public void addCheckedConstraintsEmpty(){
+ EnumMap<PrivacyLevel,LongAdder> checked = new EnumMap<>(PrivacyLevel.class);
+ CheckedConstraintsLog.addCheckedConstraints(checked);
+ assert(CheckedConstraintsLog.getCheckedConstraints() != null && CheckedConstraintsLog.getCheckedConstraints().isEmpty());
+ }
+
+ @Test
+ public void addCheckedConstraintsSingleValue(){
+ EnumMap<PrivacyLevel,LongAdder> checked = getMap(PrivacyLevel.Private, 300);
+ CheckedConstraintsLog.addCheckedConstraints(checked);
+ assert(CheckedConstraintsLog.getCheckedConstraints().get(PrivacyLevel.Private).longValue() == 300);
+ }
+
+ @Test
+ public void addCheckedConstraintsTwoValues(){
+ EnumMap<PrivacyLevel,LongAdder> checked = getMap(PrivacyLevel.Private, 300);
+ CheckedConstraintsLog.addCheckedConstraints(checked);
+ EnumMap<PrivacyLevel,LongAdder> checked2 = getMap(PrivacyLevel.Private, 150);
+ CheckedConstraintsLog.addCheckedConstraints(checked2);
+ assert(CheckedConstraintsLog.getCheckedConstraints().get(PrivacyLevel.Private).longValue() == 450);
+ }
+
+ @Test
+ public void addCheckedConstraintsMultipleValues(){
+ EnumMap<PrivacyLevel,LongAdder> checked = getMap(PrivacyLevel.Private, 300);
+ CheckedConstraintsLog.addCheckedConstraints(checked);
+ EnumMap<PrivacyLevel,LongAdder> checked2 = getMap(PrivacyLevel.Private, 150);
+ CheckedConstraintsLog.addCheckedConstraints(checked2);
+ EnumMap<PrivacyLevel,LongAdder> checked3 = getMap(PrivacyLevel.PrivateAggregation, 150);
+ CheckedConstraintsLog.addCheckedConstraints(checked3);
+ assert(CheckedConstraintsLog.getCheckedConstraints().get(PrivacyLevel.Private).longValue() == 450
+ && CheckedConstraintsLog.getCheckedConstraints().get(PrivacyLevel.PrivateAggregation).longValue() == 150);
+ }
+
+ private EnumMap<PrivacyLevel,LongAdder> getMap(PrivacyLevel level, long value){
+ EnumMap<PrivacyLevel,LongAdder> checked = new EnumMap<>(PrivacyLevel.class);
+ LongAdder valueAdder = new LongAdder();
+ valueAdder.add(value);
+ checked.put(level, valueAdder);
+ return checked;
+ }
+}
\ No newline at end of file
diff --git a/src/test/java/org/apache/sysds/test/functions/privacy/FederatedL2SVMTest.java b/src/test/java/org/apache/sysds/test/functions/privacy/FederatedL2SVMTest.java
index c93b660..f3bf331 100644
--- a/src/test/java/org/apache/sysds/test/functions/privacy/FederatedL2SVMTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/privacy/FederatedL2SVMTest.java
@@ -366,7 +366,7 @@
// Run actual dml script with federated matrix
fullDMLScriptName = HOME + TEST_NAME + ".dml";
- programArgs = new String[] {"-args", "\"localhost:" + port1 + "/" + input("X1") + "\"",
+ programArgs = new String[] {"-checkPrivacy", "-args", "\"localhost:" + port1 + "/" + input("X1") + "\"",
"\"localhost:" + port2 + "/" + input("X2") + "\"", Integer.toString(rows), Integer.toString(cols),
Integer.toString(halfRows), input("Y"), output("Z")};
runTest(true, exception2, expectedException2, -1);
@@ -374,6 +374,9 @@
if ( !(exception1 || exception2) ) {
compareResults(1e-9);
}
+
+ if ( expectedPrivacyLevel != null)
+ assert(checkedPrivacyConstraintsContains(expectedPrivacyLevel));
}
finally {
TestUtils.shutdownThreads(t1, t2);
diff --git a/src/test/java/org/apache/sysds/test/functions/privacy/FederatedWorkerHandlerTest.java b/src/test/java/org/apache/sysds/test/functions/privacy/FederatedWorkerHandlerTest.java
index f74e3a9..129b3b8 100644
--- a/src/test/java/org/apache/sysds/test/functions/privacy/FederatedWorkerHandlerTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/privacy/FederatedWorkerHandlerTest.java
@@ -33,6 +33,7 @@
import org.apache.sysds.common.Types;
import static java.lang.Thread.sleep;
+@net.jcip.annotations.NotThreadSafe
public class FederatedWorkerHandlerTest extends AutomatedTestBase {
private static final String TEST_DIR = "functions/federated/";
@@ -92,11 +93,11 @@
if (expectedException == null)
writeExpectedMatrix("R", r);
- runGenericScalarTest(TEST_PROG_SCALAR_ADDITION_MATRIX, s, expectedException);
+ runGenericScalarTest(TEST_PROG_SCALAR_ADDITION_MATRIX, s, expectedException, privacyLevel);
}
- private void runGenericScalarTest(String dmlFile, int s, Class<?> expectedException)
+ private void runGenericScalarTest(String dmlFile, int s, Class<?> expectedException, PrivacyLevel privacyLevel)
{
boolean sparkConfigOld = DMLScript.USE_LOCAL_SPARK_CONFIG;
Types.ExecMode platformOld = rtplatform;
@@ -113,7 +114,7 @@
t.start();
sleep(FED_WORKER_WAIT);
fullDMLScriptName = SCRIPT_DIR + TEST_DIR_SCALAR + dmlFile + ".dml";
- programArgs = new String[]{"-args",
+ programArgs = new String[]{"-checkPrivacy", "-args",
TestUtils.federatedAddress(FEDERATED_WORKER_HOST, FEDERATED_WORKER_PORT, input("M")),
Integer.toString(rows), Integer.toString(cols),
Integer.toString(s),
@@ -127,6 +128,7 @@
e.printStackTrace();
assert (false);
} finally {
+ assert(checkedPrivacyConstraintsContains(privacyLevel));
rtplatform = platformOld;
TestUtils.shutdownThread(t);
rtplatform = platformOld;
@@ -188,7 +190,7 @@
TestConfiguration config = availableTestConfigurations.get("aggregation");
loadTestConfiguration(config);
fullDMLScriptName = HOME + AGGREGATION_TEST_NAME + ".dml";
- programArgs = new String[] {"-args", "\"localhost:" + port + "/" + input("A") + "\"", Integer.toString(rows),
+ programArgs = new String[] {"-checkPrivacy", "-args", "\"localhost:" + port + "/" + input("A") + "\"", Integer.toString(rows),
Integer.toString(cols), Integer.toString(rows * 2), output("S"), output("R"), output("C")};
runTest(true, (expectedException != null), expectedException, -1);
@@ -197,6 +199,8 @@
if ( expectedException == null )
compareResults(1e-11);
+ assert(checkedPrivacyConstraintsContains(privacyLevel));
+
TestUtils.shutdownThread(t);
rtplatform = platformOld;
DMLScript.USE_LOCAL_SPARK_CONFIG = sparkConfigOld;
@@ -236,7 +240,7 @@
rtplatform = Types.ExecMode.SINGLE_NODE;
// Run reference dml script with normal matrix for Row/Col sum
fullDMLScriptName = HOME + TRANSFER_TEST_NAME + "Reference.dml";
- programArgs = new String[] {"-args", input("A"), expected("R"), expected("C")};
+ programArgs = new String[] {"-checkPrivacy", "-args", input("A"), expected("R"), expected("C")};
runTest(true, false, null, -1);
// reference file should not be written to hdfs, so we set platform here
@@ -247,7 +251,7 @@
TestConfiguration config = availableTestConfigurations.get("transfer");
loadTestConfiguration(config);
fullDMLScriptName = HOME + TRANSFER_TEST_NAME + ".dml";
- programArgs = new String[] {"-args", "\"localhost:" + port + "/" + input("A") + "\"", Integer.toString(rows),
+ programArgs = new String[] {"-checkPrivacy", "-args", "\"localhost:" + port + "/" + input("A") + "\"", Integer.toString(rows),
Integer.toString(cols), output("R"), output("C")};
runTest(true, (expectedException != null), expectedException, -1);
@@ -255,6 +259,8 @@
// compare all sums via files
if ( expectedException == null )
compareResults(1e-11);
+
+ assert(checkedPrivacyConstraintsContains(privacyLevel));
TestUtils.shutdownThread(t);
rtplatform = platformOld;
@@ -319,7 +325,8 @@
// Run actual dml script with federated matrix
fullDMLScriptName = HOME + MATVECMULT_TEST_NAME + ".dml";
- programArgs = new String[] {"-nvargs",
+ programArgs = new String[] {"-checkPrivacy",
+ "-nvargs",
"X1=" + TestUtils.federatedAddress("localhost", port1, input("X1")),
"X2=" + TestUtils.federatedAddress("localhost", port2, input("X2")),
"Y1=" + TestUtils.federatedAddress("localhost", port1, input("Y1")),
@@ -331,6 +338,8 @@
if (expectedException == null)
compareResults(1e-9);
+ assert(checkedPrivacyConstraintsContains(privacyLevel));
+
TestUtils.shutdownThreads(t1, t2);
rtplatform = platformOld;
diff --git a/src/test/java/org/apache/sysds/test/functions/privacy/GLMTest.java b/src/test/java/org/apache/sysds/test/functions/privacy/GLMTest.java
new file mode 100644
index 0000000..271a711
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/functions/privacy/GLMTest.java
@@ -0,0 +1,424 @@
+/*
+ * 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.test.functions.privacy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.apache.sysds.api.DMLException;
+import org.apache.sysds.hops.OptimizerUtils;
+import org.apache.sysds.runtime.matrix.data.MatrixValue.CellIndex;
+import org.apache.sysds.runtime.meta.MatrixCharacteristics;
+import org.apache.sysds.runtime.privacy.PrivacyConstraint;
+import org.apache.sysds.runtime.privacy.PrivacyConstraint.PrivacyLevel;
+import org.apache.sysds.test.AutomatedTestBase;
+import org.apache.sysds.test.TestUtils;
+
+@RunWith(value = Parameterized.class)
+@net.jcip.annotations.NotThreadSafe
+public class GLMTest extends AutomatedTestBase
+{
+ protected final static String TEST_DIR = "applications/glm/";
+ protected final static String TEST_NAME = "GLM";
+ protected String TEST_CLASS_DIR = TEST_DIR + GLMTest.class.getSimpleName() + "/";
+
+ protected int numRecords, numFeatures, distFamilyType, linkType;
+ protected double distParam, linkPower, intercept, logFeatureVarianceDisbalance, avgLinearForm, stdevLinearForm, dispersion;
+
+ protected GLMType glmType;
+
+ public enum GLMType {
+ Gaussianlog,
+ Gaussianid,
+ Gaussianinverse,
+ Poissonlog1,
+ Poissonlog2 ,
+ Poissonsqrt,
+ Poissonid,
+ Gammalog,
+ Gammainverse,
+ InvGaussian1mu,
+ InvGaussianinverse,
+ InvGaussianlog,
+ InvGaussianid,
+
+ Bernoullilog,
+ Bernoulliid,
+ Bernoullisqrt,
+ Bernoullilogit1,
+ Bernoullilogit2,
+ Bernoulliprobit1,
+ Bernoulliprobit2,
+ Bernoullicloglog1,
+ Bernoullicloglog2,
+ Bernoullicauchit,
+ Binomiallog,
+ Binomialid,
+ Binomialsqrt,
+ Binomiallogit,
+ Binomialprobit,
+ Binomialcloglog,
+ Binomialcauchit
+ }
+
+ public GLMTest (int numRecords_, int numFeatures_, int distFamilyType_, double distParam_,
+ int linkType_, double linkPower_, double intercept_, double logFeatureVarianceDisbalance_,
+ double avgLinearForm_, double stdevLinearForm_, double dispersion_, GLMType glmType)
+ {
+ this.numRecords = numRecords_;
+ this.numFeatures = numFeatures_;
+ this.distFamilyType = distFamilyType_;
+ this.distParam = distParam_;
+ this.linkType = linkType_;
+ this.linkPower = linkPower_;
+ this.intercept = intercept_;
+ this.logFeatureVarianceDisbalance = logFeatureVarianceDisbalance_;
+ this.avgLinearForm = avgLinearForm_;
+ this.stdevLinearForm = stdevLinearForm_;
+ this.dispersion = dispersion_;
+ this.glmType = glmType;
+ }
+
+ // SUPPORTED GLM DISTRIBUTION FAMILIES AND LINKS:
+ // -----------------------------------------------
+ // INPUT PARAMETERS: MEANING: Cano-
+ // dfam vpow link lpow Distribution.link nical?
+ // -----------------------------------------------
+ // 1 0.0 1 -1.0 Gaussian.inverse
+ // 1 0.0 1 0.0 Gaussian.log
+ // 1 0.0 1 1.0 Gaussian.id Yes
+ // 1 1.0 1 0.0 Poisson.log Yes
+ // 1 1.0 1 0.5 Poisson.sqrt
+ // 1 1.0 1 1.0 Poisson.id
+ // 1 2.0 1 -1.0 Gamma.inverse Yes
+ // 1 2.0 1 0.0 Gamma.log
+ // 1 2.0 1 1.0 Gamma.id
+ // 1 3.0 1 -2.0 InvGaussian.1/mu^2 Yes
+ // 1 3.0 1 -1.0 InvGaussian.inverse
+ // 1 3.0 1 0.0 InvGaussian.log
+ // 1 3.0 1 1.0 InvGaussian.id
+ // 1 * 1 * AnyVariance.AnyLink
+ // -----------------------------------------------
+ // 2 * 1 0.0 Binomial.log
+ // 2 * 2 * Binomial.logit Yes
+ // 2 * 3 * Binomial.probit
+ // 2 * 4 * Binomial.cloglog
+ // 2 * 5 * Binomial.cauchit
+ // -----------------------------------------------
+
+ @Parameters
+ public static Collection<Object[]> data() {
+ // SCHEMA:
+ // #RECORDS, #FEATURES, DISTRIBUTION_FAMILY, VARIANCE_POWER or BERNOULLI_NO, LINK_TYPE, LINK_POWER,
+ // INTERCEPT, LOG_FEATURE_VARIANCE_DISBALANCE, AVG_LINEAR_FORM, ST_DEV_LINEAR_FORM, DISPERSION, GLMTYPE
+ Object[][] data = new Object[][] {
+
+ // THIS IS TO TEST "INTERCEPT AND SHIFT/SCALE" OPTION ("icpt=2"):
+ { 200000, 50, 1, 0.0, 1, 0.0, 0.01, 3.0, 10.0, 2.0, 2.5, GLMType.Gaussianlog }, // Gaussian.log // CHECK DEVIANCE !!!
+ { 10000, 100, 1, 0.0, 1, 1.0, 0.01, 3.0, 0.0, 2.0, 2.5, GLMType.Gaussianid }, // Gaussian.id
+ { 20000, 100, 1, 0.0, 1, -1.0, 0.01, 0.0, 0.2, 0.03, 2.5, GLMType.Gaussianinverse }, // Gaussian.inverse
+ { 10000, 100, 1, 1.0, 1, 0.0, 0.01, 3.0, 0.0, 1.0, 2.5, GLMType.Poissonlog1 }, // Poisson.log
+ { 100000, 10, 1, 1.0, 1, 0.0, 0.01, 3.0, 0.0, 50.0, 2.5, GLMType.Poissonlog2 }, // Poisson.log // Pr[0|x] gets near 1
+ { 20000, 100, 1, 1.0, 1, 0.5, 0.01, 3.0, 10.0, 2.0, 2.5, GLMType.Poissonsqrt }, // Poisson.sqrt
+ { 10000, 100, 1, 1.0, 1, 1.0, 0.01, 3.0, 50.0, 10.0, 2.5, GLMType.Poissonid }, // Poisson.id
+ { 50000, 100, 1, 2.0, 1, 0.0, 0.01, 3.0, 0.0, 2.0, 2.5, GLMType.Gammalog }, // Gamma.log
+ { 10000, 100, 1, 2.0, 1, -1.0, 0.01, 3.0, 2.0, 0.3, 2.0, GLMType.Gammainverse }, // Gamma.inverse
+ { 10000, 100, 1, 3.0, 1, -2.0, 1.0, 3.0, 50.0, 7.0, 1.7, GLMType.InvGaussian1mu }, // InvGaussian.1/mu^2
+ { 10000, 100, 1, 3.0, 1, -1.0, 0.01, 3.0, 10.0, 2.0, 2.5, GLMType.InvGaussianinverse },// InvGaussian.inverse
+ { 100000, 50, 1, 3.0, 1, 0.0, 0.5, 3.0, -2.0, 1.0, 2.5, GLMType.InvGaussianlog }, // InvGaussian.log
+ { 100000, 100, 1, 3.0, 1, 1.0, 0.01, 3.0, 0.2, 0.03, 2.5, GLMType.InvGaussianid }, // InvGaussian.id
+
+ { 100000, 50, 2, -1.0, 1, 0.0, 0.01, 3.0, -5.0, 1.0, 1.0, GLMType.Bernoullilog }, // Bernoulli {-1, 1}.log // Note: Y is sparse
+ { 100000, 50, 2, -1.0, 1, 1.0, 0.01, 3.0, 0.4, 0.1, 1.0, GLMType.Bernoulliid }, // Bernoulli {-1, 1}.id
+ { 100000, 40, 2, -1.0, 1, 0.5, 0.1, 3.0, 0.4, 0.1, 1.0, GLMType.Bernoullisqrt }, // Bernoulli {-1, 1}.sqrt
+ { 10000, 100, 2, -1.0, 2, 0.0, 0.01, 3.0, 0.0, 2.0, 1.0, GLMType.Bernoullilogit1 }, // Bernoulli {-1, 1}.logit
+ { 10000, 100, 2, -1.0, 2, 0.0, 0.01, 3.0, 0.0, 50.0, 1.0, GLMType.Bernoullilogit2 }, // Bernoulli {-1, 1}.logit // Pr[y|x] near 0, 1
+ { 20000, 100, 2, -1.0, 3, 0.0, 0.01, 3.0, 0.0, 2.0, 1.0, GLMType.Bernoulliprobit1 }, // Bernoulli {-1, 1}.probit
+ { 100000, 10, 2, -1.0, 3, 0.0, 0.01, 3.0, 0.0, 50.0, 1.0, GLMType.Bernoulliprobit2 }, // Bernoulli {-1, 1}.probit // Pr[y|x] near 0, 1
+ { 10000, 100, 2, -1.0, 4, 0.0, 0.01, 3.0, -2.0, 1.0, 1.0, GLMType.Bernoullicloglog1 }, // Bernoulli {-1, 1}.cloglog
+ { 50000, 20, 2, -1.0, 4, 0.0, 0.01, 3.0, -2.0, 50.0, 1.0, GLMType.Bernoullicloglog2 }, // Bernoulli {-1, 1}.cloglog // Pr[y|x] near 0, 1
+ { 20000, 100, 2, -1.0, 5, 0.0, 0.01, 3.0, 0.0, 2.0, 1.0, GLMType.Bernoullicauchit }, // Bernoulli {-1, 1}.cauchit
+
+ { 50000, 100, 2, 1.0, 1, 0.0, 0.01, 3.0, -5.0, 1.0, 2.5, GLMType.Binomiallog }, // Binomial two-column.log // Note: Y is sparse
+ { 10000, 100, 2, 1.0, 1, 1.0, 0.0, 0.0, 0.4, 0.05, 2.5, GLMType.Binomialid }, // Binomial two-column.id
+ { 100000, 100, 2, 1.0, 1, 0.5, 0.1, 3.0, 0.4, 0.05, 2.5, GLMType.Binomialsqrt }, // Binomial two-column.sqrt
+ { 10000, 100, 2, 1.0, 2, 0.0, 0.01, 3.0, 0.0, 2.0, 2.5, GLMType.Binomiallogit }, // Binomial two-column.logit
+ { 20000, 100, 2, 1.0, 3, 0.0, 0.01, 3.0, 0.0, 2.0, 2.5, GLMType.Binomialprobit }, // Binomial two-column.probit
+ { 10000, 100, 2, 1.0, 4, 0.0, 0.01, 3.0, -2.0, 1.0, 2.5, GLMType.Binomialcloglog }, // Binomial two-column.cloglog
+ { 20000, 100, 2, 1.0, 5, 0.0, 0.01, 3.0, 0.0, 2.0, 2.5, GLMType.Binomialcauchit }, // Binomial two-column.cauchit
+ };
+ return Arrays.asList(data);
+ }
+
+ @Override
+ public void setUp()
+ {
+ addTestConfiguration(TEST_CLASS_DIR, TEST_NAME);
+ }
+
+ @Test
+ public void TestGLMPrivateX(){
+
+ PrivacyConstraint pc = new PrivacyConstraint(PrivacyLevel.Private);
+ Class<?> expectedException = null;
+ switch ( glmType ){
+ case Gaussianinverse:
+ case Poissonlog1:
+ case Poissonlog2:
+ case Poissonsqrt:
+ case Poissonid:
+ case Gammalog:
+ case Gammainverse:
+ case InvGaussian1mu:
+ case InvGaussianinverse:
+ case InvGaussianlog:
+ case InvGaussianid:
+ case Binomialid:
+ case Binomialcauchit:
+ case Gaussianlog:
+ case Gaussianid:
+ case Bernoullilog:
+ case Bernoulliid:
+ case Bernoullisqrt:
+ case Bernoullilogit1:
+ case Bernoullilogit2:
+ case Bernoulliprobit1:
+ case Bernoulliprobit2:
+ case Bernoullicloglog1:
+ case Bernoullicloglog2:
+ case Bernoullicauchit:
+ case Binomiallog:
+ case Binomialsqrt:
+ case Binomiallogit:
+ case Binomialprobit:
+ case Binomialcloglog:
+ expectedException = DMLException.class;
+ break;
+ default:
+ expectedException = null;
+ break;
+ }
+ testGLM(pc, null, expectedException);
+ }
+
+ @Test
+ public void TestGLMPrivateAggregationX(){
+ PrivacyConstraint pc = new PrivacyConstraint(PrivacyLevel.PrivateAggregation);
+ Class<?> expectedException = null;
+ testGLM(pc, null, expectedException);
+ }
+
+ @Test
+ public void TestGLMNonePrivateX(){
+ PrivacyConstraint pc = new PrivacyConstraint(PrivacyLevel.None);
+ Class<?> expectedException = null;
+ testGLM(pc, null, expectedException);
+ }
+
+ @Test
+ public void TestGLMPrivateY(){
+ PrivacyConstraint pc = new PrivacyConstraint(PrivacyLevel.Private);
+ Class<?> expectedException = null;
+ switch ( glmType ){
+ case Gaussianinverse:
+ case Poissonlog1:
+ case Poissonlog2:
+ case Poissonsqrt:
+ case Poissonid:
+ case Gammalog:
+ case Gammainverse:
+ case InvGaussian1mu:
+ case InvGaussianinverse:
+ case InvGaussianlog:
+ case InvGaussianid:
+ case Binomialid:
+ case Binomialcauchit:
+ case Gaussianlog:
+ case Gaussianid:
+ case Bernoullilog:
+ case Bernoulliid:
+ case Bernoullisqrt:
+ case Bernoullilogit1:
+ case Bernoullilogit2:
+ case Bernoulliprobit1:
+ case Bernoulliprobit2:
+ case Bernoullicloglog1:
+ case Bernoullicloglog2:
+ case Bernoullicauchit:
+ case Binomiallog:
+ case Binomialsqrt:
+ case Binomiallogit:
+ case Binomialprobit:
+ case Binomialcloglog:
+ expectedException = DMLException.class;
+ break;
+ default:
+ expectedException = null;
+ break;
+ }
+ testGLM(null, pc, expectedException);
+ }
+
+ @Test
+ public void TestGLMPrivateAggregationY(){
+ PrivacyConstraint pc = new PrivacyConstraint(PrivacyLevel.PrivateAggregation);
+ Class<?> expectedException = null;
+ testGLM(null, pc, expectedException);
+ }
+
+ @Test
+ public void TestGLMNonePrivateY(){
+ PrivacyConstraint pc = new PrivacyConstraint(PrivacyLevel.None);
+ Class<?> expectedException = null;
+ testGLM(null, pc, expectedException);
+ }
+
+ @Test
+ public void TestGLMPrivateXY(){
+ PrivacyConstraint pc = new PrivacyConstraint(PrivacyLevel.Private);
+ testGLM(pc, pc, DMLException.class);
+ }
+
+ @Test
+ public void TestGLMPrivateAggregationXY(){
+ PrivacyConstraint pc = new PrivacyConstraint(PrivacyLevel.PrivateAggregation);
+ Class<?> expectedException = null;
+ testGLM(pc, pc, expectedException);
+ }
+
+ @Test
+ public void TestGLMNonePrivateXY(){
+ PrivacyConstraint pc = new PrivacyConstraint(PrivacyLevel.Private);
+ testGLM(pc, pc, DMLException.class);
+ }
+
+ public void testGLM(PrivacyConstraint privacyX, PrivacyConstraint privacyY, Class<?> expectedException)
+ {
+ System.out.println("------------ BEGIN " + TEST_NAME + " TEST WITH {" +
+ numRecords + ", " +
+ numFeatures + ", " +
+ distFamilyType + ", " +
+ distParam + ", " +
+ linkType + ", " +
+ linkPower + ", " +
+ intercept + ", " +
+ logFeatureVarianceDisbalance + ", " +
+ avgLinearForm + ", " +
+ stdevLinearForm + ", " +
+ dispersion +
+ "} ------------");
+ System.out.println("GLMType: " + this.glmType);
+ System.out.println(expectedException);
+
+ int rows = numRecords; // # of rows in the training data
+ int cols = numFeatures; // # of features in the training data
+
+ TestUtils.GLMDist glmdist = new TestUtils.GLMDist (distFamilyType, distParam, linkType, linkPower);
+ glmdist.set_dispersion (dispersion);
+
+ getAndLoadTestConfiguration(TEST_NAME);
+
+ // prepare training data set
+ Random r = new Random(314159265);
+ double[][] X = TestUtils.generateUnbalancedGLMInputDataX(rows, cols, logFeatureVarianceDisbalance);
+ double[] beta = TestUtils.generateUnbalancedGLMInputDataB(X, cols, intercept, avgLinearForm, stdevLinearForm, r);
+ double[][] y = TestUtils.generateUnbalancedGLMInputDataY(X, beta, rows, cols, glmdist, intercept, dispersion, r);
+
+ int defaultBlockSize = OptimizerUtils.DEFAULT_BLOCKSIZE;
+
+ MatrixCharacteristics mc_X = new MatrixCharacteristics(rows, cols, defaultBlockSize, -1);
+ writeInputMatrixWithMTD ("X", X, true, mc_X, privacyX);
+
+ MatrixCharacteristics mc_y = new MatrixCharacteristics(rows, y[0].length, defaultBlockSize, -1);
+ writeInputMatrixWithMTD ("Y", y, true, mc_y, privacyY);
+
+ List<String> proArgs = new ArrayList<>();
+ proArgs.add("-nvargs");
+ proArgs.add("dfam=" + String.format ("%d", distFamilyType));
+ proArgs.add(((distFamilyType == 2 && distParam != 1.0) ? "yneg=" : "vpow=") + String.format ("%f", distParam));
+ proArgs.add((distFamilyType == 2 && distParam != 1.0) ? "vpow=0.0" : "yneg=0.0");
+ proArgs.add("link=" + String.format ("%d", linkType));
+ proArgs.add("lpow=" + String.format ("%f", linkPower));
+ proArgs.add("icpt=2"); // INTERCEPT - CHANGE THIS AS NEEDED
+ proArgs.add("disp=0.0"); // DISPERSION (0.0: ESTIMATE)
+ proArgs.add("reg=0.0"); // LAMBDA REGULARIZER
+ proArgs.add("tol=0.000000000001"); // TOLERANCE (EPSILON)
+ proArgs.add("moi=300");
+ proArgs.add("mii=0");
+ proArgs.add("X=" + input("X"));
+ proArgs.add("Y=" + input("Y"));
+ proArgs.add("B=" + output("betas_SYSTEMDS"));
+ programArgs = proArgs.toArray(new String[proArgs.size()]);
+
+ fullDMLScriptName = "scripts/algorithms/GLM.dml";
+
+ rCmd = getRCmd(input("X.mtx"), input("Y.mtx"), String.format ("%d", distFamilyType), String.format ("%f", distParam),
+ String.format ("%d", linkType), String.format ("%f", linkPower), "1" /*intercept*/, "0.000000000001" /*tolerance (espilon)*/,
+ expected("betas_R"));
+
+ int expectedNumberOfJobs = -1; // 31;
+
+ runTest(true, (expectedException != null), expectedException, expectedNumberOfJobs);
+
+ if ( expectedException == null){
+ double max_abs_beta = 0.0;
+ HashMap<CellIndex, Double> wTRUE = new HashMap<> ();
+ for (int j = 0; j < cols; j ++)
+ {
+ wTRUE.put (new CellIndex (j + 1, 1), Double.valueOf(beta [j]));
+ max_abs_beta = (max_abs_beta >= Math.abs (beta[j]) ? max_abs_beta : Math.abs (beta[j]));
+ }
+
+ HashMap<CellIndex, Double> wSYSTEMDS_raw = readDMLMatrixFromHDFS ("betas_SYSTEMDS");
+ HashMap<CellIndex, Double> wSYSTEMDS = new HashMap<> ();
+ for (CellIndex key : wSYSTEMDS_raw.keySet())
+ if (key.column == 1)
+ wSYSTEMDS.put (key, wSYSTEMDS_raw.get (key));
+
+ runRScript(true);
+
+ HashMap<CellIndex, Double> wR = readRMatrixFromFS ("betas_R");
+
+ double eps = 0.000001;
+ if( (distParam==0 && linkType==1) ) { // Gaussian.*
+ //NOTE MB: Gaussian.log was the only test failing when we introduced multi-threaded
+ //matrix multplications (mmchain). After discussions with Sasha, we decided to change the eps
+ //because accuracy is anyway affected by various rewrites like binary to unary (-1*x->-x),
+ //transpose-matrixmult, and dot product sum. Disabling these rewrites led to a successful
+ //test result. Even without multi-threaded matrix mult this test was failing for different number
+ //of rows if these rewrites are enabled. Users can turn off rewrites if high accuracy is required.
+ //However, in the future we might also consider to use Kahan plus for aggregations in matrix mult
+ //(at least for the final aggregation of partial results from individual threads).
+
+ //NOTE MB: similar issues occurred with other tests when moving to github action tests
+ eps *= (linkPower==-1) ? 4 : 2; //Gaussian.inverse vs Gaussian.*;
+ }
+ TestUtils.compareMatrices (wR, wSYSTEMDS, eps * max_abs_beta, "wR", "wSYSTEMDS");
+ }
+ }
+}
