This is an automated email from the ASF dual-hosted git repository.
baunsgaard pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/systemds.git
The following commit(s) were added to refs/heads/master by this push:
new c9a02a2 [MINOR] Negative test of federatedError
c9a02a2 is described below
commit c9a02a2e9fa3ec50a2a8c3611d62a2a334b5bab3
Author: baunsgaard <[email protected]>
AuthorDate: Thu Aug 20 19:48:46 2020 +0200
[MINOR] Negative test of federatedError
---
.../federated/FederatedResponse.java | 6 +-
.../federated/FederatedWorkerHandler.java | 2 +-
.../org/apache/sysds/test/AutomatedTestBase.java | 22 +++++-
.../functions/federated/FederatedNegativeTest.java | 78 ++++++++++++++++++++++
4 files changed, 105 insertions(+), 3 deletions(-)
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 ea03bd4..5b0e4eb 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
@@ -91,7 +91,11 @@ public class FederatedResponse implements Serializable {
throw (Exception) potentialException;
}
}
- throw new DMLRuntimeException("Unknown runtime exception in
handling of federated request by federated worker.");
+ String errorMessage = getErrorMessage();
+ if (getErrorMessage() != "No readable error message")
+ throw new DMLRuntimeException(errorMessage);
+ else
+ throw new DMLRuntimeException("Unknown runtime
exception in handling of federated request by federated worker.");
}
/**
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 b0c75df..35b844d 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
@@ -295,7 +295,7 @@ public class FederatedWorkerHandler extends
ChannelInboundHandlerAdapter {
@Override
public void operationComplete(ChannelFuture channelFuture)
throws InterruptedException {
if (!channelFuture.isSuccess()){
- log.fatal("Federated Worker Write failed");
+ log.error("Federated Worker Write failed");
channelFuture
.channel()
.writeAndFlush(
diff --git a/src/test/java/org/apache/sysds/test/AutomatedTestBase.java
b/src/test/java/org/apache/sysds/test/AutomatedTestBase.java
index b40a637..bfb2e5d 100644
--- a/src/test/java/org/apache/sysds/test/AutomatedTestBase.java
+++ b/src/test/java/org/apache/sysds/test/AutomatedTestBase.java
@@ -1273,7 +1273,7 @@ public abstract class AutomatedTestBase {
args.add("-gpu");
}
- protected int getRandomAvailablePort() {
+ public static int getRandomAvailablePort() {
try(ServerSocket availableSocket = new ServerSocket(0)) {
return availableSocket.getLocalPort();
}
@@ -1312,6 +1312,26 @@ public abstract class AutomatedTestBase {
return t;
}
+ public static Thread startLocalFedWorkerWithArgs(String[] args) {
+ Thread t = null;
+
+ try {
+ t = new Thread(() -> {
+ try {
+ main(args);
+ }
+ catch(IOException e) {
+ }
+ });
+ t.start();
+
java.util.concurrent.TimeUnit.MILLISECONDS.sleep(FED_WORKER_WAIT);
+ }
+ catch(InterruptedException e) {
+ // Should happen at closing of the worker so don't print
+ }
+ return t;
+ }
+
private boolean rCompareException(boolean exceptionExpected, String
errMessage, Throwable e, boolean result) {
if(e.getCause() != null) {
result |= rCompareException(exceptionExpected,
errMessage, e.getCause(), result);
diff --git
a/src/test/java/org/apache/sysds/test/functions/federated/FederatedNegativeTest.java
b/src/test/java/org/apache/sysds/test/functions/federated/FederatedNegativeTest.java
new file mode 100644
index 0000000..a355275
--- /dev/null
+++
b/src/test/java/org/apache/sysds/test/functions/federated/FederatedNegativeTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.federated;
+
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.sysds.common.Types;
+import org.apache.sysds.runtime.controlprogram.federated.*;
+import org.apache.sysds.test.AutomatedTestBase;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Test;
+
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
[email protected]
+public class FederatedNegativeTest {
+ protected static Logger log =
Logger.getLogger(FederatedNegativeTest.class);
+
+ static {
+ Logger.getLogger("org.apache.sysds").setLevel(Level.OFF);
+ }
+
+ @Test
+ public void NegativeTest1() {
+ int port = AutomatedTestBase.getRandomAvailablePort();
+ String[] args = {"-w", Integer.toString(port)};
+ Thread t = AutomatedTestBase.startLocalFedWorkerWithArgs(args);
+ Map<FederatedRange, FederatedData> fedMap = new HashMap<>();
+ FederatedRange r = new FederatedRange(new long[]{0,0}, new
long[]{1,1});
+ FederatedData d = new FederatedData(
+ Types.DataType.SCALAR,
+ new InetSocketAddress("localhost", port),
+ "Nowhere");
+ fedMap.put(r,d);
+ FederationMap fedM = new FederationMap(fedMap);
+ FederatedRequest fr = new
FederatedRequest(FederatedRequest.RequestType.GET_VAR);
+ Future<FederatedResponse>[] res = fedM.execute(0, fr);
+ try {
+ FederatedResponse fres = res[0].get();
+ assertFalse(fres.isSuccessful());
+ assertTrue(fres.getErrorMessage().contains("Variable 0
does not exist at federated worker"));
+
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ } catch (ExecutionException e) {
+ e.printStackTrace();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ TestUtils.shutdownThread(t);
+ }
+
+}
