diff --git a/Exareme-Docker/files/root/exareme/bootstrap.sh b/Exareme-Docker/files/root/exareme/bootstrap.sh
index 62c2c53cf3a712606f0ca0c986d129e7ae873fc1..46ed522bd43949d0038476675db0346557ea4025 100755
--- a/Exareme-Docker/files/root/exareme/bootstrap.sh
+++ b/Exareme-Docker/files/root/exareme/bootstrap.sh
@@ -14,7 +14,7 @@ CONSUL_WAIT_FOR_MASTER_IP_MAX_ATTEMPTS=20
EXAREME_NODE_STARTUP_HEALTH_CHECK_MAX_ATTEMPTS=10
EXAREME_NODE_HEALTH_CHECK_TIMEOUT=60
MASTER_NODE_REACHABLE_TIMEOUT=5
-PERIODIC_EXAREME_NODES_HEALTH_CHECK_MAX_RETRIES=3
+PERIODIC_EXAREME_NODES_HEALTH_CHECK_MAX_RETRIES=5
PERIODIC_EXAREME_NODES_HEALTH_CHECK_INTERVAL=120
EXAREME_HEALTH_CHECK_AWAIT_TIME=60
PERIODIC_TEMP_FILES_REMOVAL=300
@@ -118,7 +118,7 @@ exaremeNodesHealthCheck() {
echo "$(timestamp) HEALTH CHECK for node with IP ${NODE_IP} and name ${NODE_NAME} ."
if [[ "${FEDERATION_ROLE}" == "master" ]]; then
- check=$(curl -s --max-time ${EXAREME_NODE_HEALTH_CHECK_TIMEOUT} "${NODE_IP}:9092/check/worker?NODE_IP=${NODE_IP}&NODE_NAME=${NODE_NAME}")
+ check=$(curl -s -X POST --max-time ${EXAREME_NODE_HEALTH_CHECK_TIMEOUT} ${NODE_IP}:9090/mining/query/HEALTH_CHECK)
else
check=$(curl -s --max-time ${EXAREME_NODE_HEALTH_CHECK_TIMEOUT} "${MASTER_IP}:9092/check/worker?NODE_IP=${NODE_IP}&NODE_NAME=${NODE_NAME}")
fi
@@ -255,11 +255,6 @@ if [[ "${FEDERATION_ROLE}" == "master" ]]; then
curl -s -X PUT -d @- ${CONSULURL}/v1/kv/${CONSUL_MASTER_PATH}/${NODE_NAME} <<<${NODE_IP}
curl -s -X PUT -d @- ${CONSULURL}/v1/kv/${CONSUL_ACTIVE_WORKERS_PATH}/${NODE_NAME} <<<${NODE_IP}
- if ! startupExaremeNodesHealthCheck; then
- echo "$(timestamp) HEALTH CHECK algorithm failed. Switch ENVIRONMENT_TYPE to 'DEV' to see error messages coming from EXAREME. Exiting..."
- exit 1
- fi
-
periodicExaremeNodesHealthCheck &
else ##### Running bootstrap on a worker node #####
@@ -271,16 +266,16 @@ else ##### Running bootstrap on a worker node #####
echo "$(timestamp) Starting Exareme on worker node with IP: ${NODE_IP} and nodeName: ${NODE_NAME}"
. ./start-worker.sh
-
- # Updating consul with node IP
- echo -e "\n$(timestamp) Updating consul with worker node IP."
- curl -s -X PUT -d @- ${CONSULURL}/v1/kv/${CONSUL_ACTIVE_WORKERS_PATH}/${NODE_NAME} <<<${NODE_IP}
-
+
if ! startupExaremeNodesHealthCheck; then
echo "$(timestamp) HEALTH CHECK algorithm failed. Switch ENVIRONMENT_TYPE to 'DEV' to see error messages coming from EXAREME. Exiting..."
exit 1
fi
+ # Updating consul with node IP
+ echo -e "\n$(timestamp) Updating consul with worker node IP."
+ curl -s -X PUT -d @- ${CONSULURL}/v1/kv/${CONSUL_ACTIVE_WORKERS_PATH}/${NODE_NAME} <<<${NODE_IP}
+
periodicExaremeNodesHealthCheck &
periodicReachableMasterNodeCheck &
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/client/AdpDBClientQueryStatus.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/client/AdpDBClientQueryStatus.java
index 14c9f40019b7b52ce5d8ffb1286d11f894c9c1be..4010a06f1e7aa9d903bd8a87960293b2e62c1175 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/client/AdpDBClientQueryStatus.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/client/AdpDBClientQueryStatus.java
@@ -28,8 +28,8 @@ public interface AdpDBClientQueryStatus {
void registerListener(AdpDBQueryListener listener) throws RemoteException;
- InputStream getResult() throws RemoteException;
+ String getResult() throws RemoteException;
- InputStream getResult(DataSerialization ds) throws RemoteException;
+ String getResult(DataSerialization ds) throws RemoteException;
}
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/client/rmi/RmiAdpDBClientQueryStatus.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/client/rmi/RmiAdpDBClientQueryStatus.java
index a692c7a8d02cd35d8395baaa09d21e2f2bbad019..2e627f04e8a571558c10ac1c72991756316c20fe 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/client/rmi/RmiAdpDBClientQueryStatus.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/client/rmi/RmiAdpDBClientQueryStatus.java
@@ -21,7 +21,9 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.rmi.RemoteException;
+import java.util.Arrays;
import java.util.HashMap;
+import java.util.concurrent.FutureTask;
/**
* @author alex
@@ -35,7 +37,8 @@ public class RmiAdpDBClientQueryStatus implements AdpDBClientQueryStatus {
private String lastStatus;
private TimeFormat timeF;
private boolean finished;
- private InputStream result;
+ private boolean error;
+ private String result;
public RmiAdpDBClientQueryStatus(AdpDBQueryID queryId, AdpDBClientProperties properties,
AdpDBQueryExecutionPlan plan, AdpDBStatus status) {
@@ -45,6 +48,7 @@ public class RmiAdpDBClientQueryStatus implements AdpDBClientQueryStatus {
this.lastStatus = null;
this.timeF = new TimeFormat(TimeUnit.min);
this.finished = false;
+ this.error = false;
result = null;
}
@@ -61,13 +65,6 @@ public class RmiAdpDBClientQueryStatus implements AdpDBClientQueryStatus {
if (!status.hasFinished() && !status.hasError())
return false;
- try {
- String algorithmResult = IOUtils.toString(getResult(DataSerialization.summary), StandardCharsets.UTF_8);
- log.info("Algorithm with queryId " + getQueryID().getQueryID() + " terminated. Result: \n " + algorithmResult);
- } catch (IOException e) {
- log.error("Could not read the algorithm result table." + getQueryID());
- }
-
finished = true;
return true;
}
@@ -113,7 +110,7 @@ public class RmiAdpDBClientQueryStatus implements AdpDBClientQueryStatus {
}
@Override
- public InputStream getResult() throws RemoteException {
+ public String getResult() throws RemoteException {
return getResult(DataSerialization.ldjson);
}
@@ -129,14 +126,27 @@ public class RmiAdpDBClientQueryStatus implements AdpDBClientQueryStatus {
* @throws RemoteException
*/
@Override
- public InputStream getResult(DataSerialization ds) throws RemoteException {
+ public String getResult(DataSerialization ds) throws RemoteException {
// The registry should be updated the 1st time we fetch a result stream.
if (result == null) {
updateRegistry();
}
- result = new RmiAdpDBClient(AdpDBManagerLocator.getDBManager(), properties)
+ InputStream resultStream = new RmiAdpDBClient(AdpDBManagerLocator.getDBManager(), properties)
.readTable(plan.getResultTables().get(0).getName(), ds);
+
+ FutureTask<String> getResultFromStream;
+ try {
+ getResultFromStream = new FutureTask<>(() ->
+ IOUtils.toString(resultStream, StandardCharsets.UTF_8));
+
+ new Thread(getResultFromStream).start();
+ result = getResultFromStream.get(30, java.util.concurrent.TimeUnit.SECONDS);
+ } catch (Exception e) {
+ log.error("Error reading the result table! QueryID:" + status.getQueryID().getQueryID(), e);
+ throw new RemoteException("Could not read the result table!");
+ }
+
return result;
}
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/executor/remote/AdpDBArtJobMonitor.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/executor/remote/AdpDBArtJobMonitor.java
index f08d2f57fe3b0ed0d6e7d290e20ea766b85d0d3a..393f7ea87bbd1ac08ca8864cdd716f19d8931612 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/executor/remote/AdpDBArtJobMonitor.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/executor/remote/AdpDBArtJobMonitor.java
@@ -60,10 +60,9 @@ public class AdpDBArtJobMonitor implements Runnable {
statusManager.getStatistics(status.getId()).setTotalDataTransfers(stats.getTotalData());
while (!sessionManager.hasFinished() && !sessionManager.hasError()) {
-
boolean updateProgressStatistics = updateProgressStatistics();
if (updateProgressStatistics) {
- log.info("Session is running... ID: " + sessionPlan.getSessionID().getLongId()
+ log.info("Session is updating... ID: " + sessionPlan.getSessionID().getLongId()
+ " , QueryID: " + queryID.getQueryID());
log.debug("Update listeners ...");
synchronized (listeners) {
@@ -89,13 +88,15 @@ public class AdpDBArtJobMonitor implements Runnable {
.setAdpEngineStatistics(statsManager.getStatistics());
if (sessionManager != null && !sessionManager.hasError()) {
- log.debug("Session finished, closing! ID: " + sessionPlan.getSessionID().getLongId()
+ log.info("Session finished, closing! ID: " + sessionPlan.getSessionID().getLongId()
+ " , QueryID: " + queryID.getQueryID());
statusManager.setFinished(status.getId());
} else {
+ log.info("Session error! ID: " + sessionPlan.getSessionID().getLongId()
+ + " , QueryID: " + queryID.getQueryID());
statusManager.setError(status.getId(), sessionManager.getErrorList().get(0));
}
- log.debug("Session closing! ID: "+ sessionPlan.getSessionID().getLongId()
+ log.debug("Session closing! ID: " + sessionPlan.getSessionID().getLongId()
+ " , QueryID: " + queryID.getQueryID());
sessionPlan.close();
@@ -104,7 +105,6 @@ public class AdpDBArtJobMonitor implements Runnable {
log.error("Cannot monitor job, sessionID: " + sessionPlan.getSessionID().getLongId());
log.error("Cannot monitor job, queryID: " + status.getQueryID().getQueryID(), e);
} finally {
- log.debug("Terminate listeners ( " + listeners.size() + ")...");
synchronized (listeners) {
for (AdpDBQueryListener l : listeners) {
l.terminated(queryID, status);
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/executor/remote/AdpDBExecutorRemote.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/executor/remote/AdpDBExecutorRemote.java
index b0bd57fb4264281080ee153ed7026665be9b27da..513297fe2548ce18514c903b4c4d3a2e69f346cd 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/executor/remote/AdpDBExecutorRemote.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/executor/remote/AdpDBExecutorRemote.java
@@ -149,7 +149,6 @@ public class AdpDBExecutorRemote implements AdpDBExecutor {
AdpDBArtJobMonitor monitor =
new AdpDBArtJobMonitor(sessionPlan, status, statusManager, execPlan.getQueryID());
monitors.put(execPlan.getQueryID(), monitor);
-
executor.submit(monitor);
statusArray.add(status);
return status;
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/handler/NIterativeAlgorithmResultEntity.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/handler/NIterativeAlgorithmResultEntity.java
index dde8e356e83e34582efcd9ea93bcd390c9d39e37..04d23ade2c4f6e92244e2dd20b87ab8edac644c2 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/handler/NIterativeAlgorithmResultEntity.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/handler/NIterativeAlgorithmResultEntity.java
@@ -47,11 +47,6 @@ public class NIterativeAlgorithmResultEntity extends BasicHttpEntity
this.dataSerialization = dataSerialization;
}
- private final static String user_error = new String("text/plain+user_error");
- private final static String error = new String("text/plain+error");
- private final static String warning = new String("text/plain+warning");
-
-
/**
* @param encoder is used to save the output
* @param ioctrl will be used from the iterativeAlgorithmState, when the algorithm is complete,
@@ -87,9 +82,11 @@ public class NIterativeAlgorithmResultEntity extends BasicHttpEntity
if (!finalizeQueryStatus.hasError() &&
finalizeQueryStatus.hasFinished()) {
if (channel == null) {
+ String result = iterativeAlgorithmState.getAdpDBClientQueryStatus().getResult(dataSerialization);
+ log.info("Iterative algorithm with key " + iterativeAlgorithmState.getAlgorithmKey()
+ + " terminated. Result: \n " + result);
channel = Channels.newChannel(
- iterativeAlgorithmState.getAdpDBClientQueryStatus()
- .getResult(dataSerialization));
+ new ByteArrayInputStream(result.getBytes(StandardCharsets.UTF_8)));
}
// Reading from the channel to the buffer, flip is required by the API
channel.read(buffer);
@@ -183,11 +180,13 @@ public class NIterativeAlgorithmResultEntity extends BasicHttpEntity
finalizeQueryStatus.close();
finalizeQueryStatus = null;
}
+ if (iterativeAlgorithmState != null)
+ iterativeAlgorithmState.releaseLock();
iterativeAlgorithmState = null;
}
@Override
- public void close() throws IOException {
+ public void close() {
}
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/scheduler/events/IterationsEventHandler.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/scheduler/events/IterationsEventHandler.java
index 7c28a362bb47238ef2fc43b609607d34b8eabe24..0d9c009f0cb7df8cd33cfd075c05cc8329c208d6 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/scheduler/events/IterationsEventHandler.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/scheduler/events/IterationsEventHandler.java
@@ -3,7 +3,6 @@ package madgik.exareme.master.engine.iterations.scheduler.events;
import madgik.exareme.common.app.engine.AdpDBQueryID;
import madgik.exareme.master.client.AdpDBClientQueryStatus;
import madgik.exareme.master.engine.iterations.scheduler.IterationsDispatcher;
-import madgik.exareme.master.engine.iterations.scheduler.events.phaseCompletion.PhaseCompletionEventHandler;
import madgik.exareme.master.engine.iterations.state.IterationsStateManager;
import madgik.exareme.master.engine.iterations.state.IterativeAlgorithmState;
import madgik.exareme.utils.eventProcessor.EventHandler;
@@ -56,8 +55,9 @@ public abstract class IterationsEventHandler<T extends IterationsEvent>
ias.getAlgorithmKey(),
dflScript);
- log.info("New Iterative phase: " + currentPhase);
- log.info("Executing Iterative DFL Script: \n" + dflScript);
+ log.info("New Iterative phase: " + currentPhase + " for algorithm: " + ias.getAlgorithmKey() +
+ " with queryID: " + queryStatus.getQueryID().getQueryID());
+ log.debug("Executing Iterative DFL Script: \n" + dflScript);
ias.setCurrentExecutionPhase(currentPhase);
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/scheduler/events/phaseCompletion/PhaseCompletionEventHandler.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/scheduler/events/phaseCompletion/PhaseCompletionEventHandler.java
index 785bb9bb934573939a08f3810bc52385b342d8d3..7127c098b9691e33d68002ed1057d54bd3bf296c 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/scheduler/events/phaseCompletion/PhaseCompletionEventHandler.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/iterations/scheduler/events/phaseCompletion/PhaseCompletionEventHandler.java
@@ -109,8 +109,7 @@ public class PhaseCompletionEventHandler extends IterationsEventHandler<PhaseCom
String terminationConditionResult;
try {
- InputStream previousResultStream = ias.getAdpDBClientQueryStatus().getResult();
- terminationConditionResult = IOUtils.toString(previousResultStream, StandardCharsets.UTF_8);
+ terminationConditionResult = ias.getAdpDBClientQueryStatus().getResult();
} catch (IOException e) {
throw new IterationsStateFatalException(
"Could not read the termination_condition result table.", ias.getAlgorithmKey());
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/scheduler/AdpDBQueryScheduler.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/scheduler/AdpDBQueryScheduler.java
index fd785d00cce46445f46952bb4d22b909d8a80aaf..15036ca9ea375740b6521d7a170dd52881519abe 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/scheduler/AdpDBQueryScheduler.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/scheduler/AdpDBQueryScheduler.java
@@ -29,6 +29,7 @@ import madgik.exareme.master.registry.Registry;
import madgik.exareme.utils.eventProcessor.EventProcessor;
import java.rmi.RemoteException;
+import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/statusMgr/AdpDBJobSession.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/statusMgr/AdpDBJobSession.java
index 990e4a499136ac6747152f84c97731cbd60a3733..9d8d3a67d25315f305c3c12bd662f555659e9f04 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/statusMgr/AdpDBJobSession.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/engine/statusMgr/AdpDBJobSession.java
@@ -8,17 +8,14 @@ import madgik.exareme.common.app.engine.AdpDBQueryListener;
import madgik.exareme.common.app.engine.AdpDBStatistics;
import madgik.exareme.master.engine.AdpDBManagerLocator;
import madgik.exareme.worker.art.executionEngine.session.ExecutionEngineSessionPlan;
-import org.apache.log4j.Logger;
import java.rmi.RemoteException;
-import java.util.Arrays;
import java.util.Map;
/**
* @author herald
*/
public class AdpDBJobSession {
- private static final Logger log = Logger.getLogger(AdpDBJobSession.class);
private boolean finished = false;
private boolean error = false;
private Exception exception = null;
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/HBP/HBPQueryHandler.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/HBP/HBPQueryHandler.java
index cb5281154fbcdc02851b8bab1cd08dd454088add..8778a13689f036ae423f216d55a0475dc1f550ba 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/HBP/HBPQueryHandler.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/HBP/HBPQueryHandler.java
@@ -13,8 +13,8 @@ import madgik.exareme.master.engine.iterations.handler.IterationsHandler;
import madgik.exareme.master.engine.iterations.handler.NIterativeAlgorithmResultEntity;
import madgik.exareme.master.engine.iterations.state.IterativeAlgorithmState;
import madgik.exareme.master.gateway.ExaremeGatewayUtils;
-import madgik.exareme.master.gateway.async.handler.HBP.Exceptions.RequestException;
import madgik.exareme.master.gateway.async.handler.HBP.Exceptions.BadUserInputException;
+import madgik.exareme.master.gateway.async.handler.HBP.Exceptions.RequestException;
import madgik.exareme.master.gateway.async.handler.entity.NQueryResultEntity;
import madgik.exareme.master.queryProcessor.HBP.AlgorithmProperties;
import madgik.exareme.master.queryProcessor.HBP.Algorithms;
@@ -160,7 +160,7 @@ public class HBPQueryHandler implements HttpAsyncRequestHandler<HttpRequest> {
queryStatus = dbClient.query(algorithmKey, dfl);
log.info("Algorithm " + algorithmKey + " with queryID "
- + queryStatus.getQueryID() + " execution started.");
+ + queryStatus.getQueryID().getQueryID() + " execution started.");
log.debug("DFL Script: \n " + dfl);
BasicHttpEntity entity = new NQueryResultEntity(queryStatus, ds,
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/entity/NQueryResultEntity.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/entity/NQueryResultEntity.java
index 649d6a632fe78540b47951a1659aab4da044e9f6..a0dd23fde67cad66c62ba545b79f9180990c2aa6 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/entity/NQueryResultEntity.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/entity/NQueryResultEntity.java
@@ -9,14 +9,13 @@ import org.apache.http.nio.IOControl;
import org.apache.http.nio.entity.HttpAsyncContentProducer;
import org.apache.log4j.Logger;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
-/**
- * TODO flush output before suspend
- */
public class NQueryResultEntity extends BasicHttpEntity implements HttpAsyncContentProducer {
private static final Logger log = Logger.getLogger(NQueryResultEntity.class);
@@ -55,7 +54,11 @@ public class NQueryResultEntity extends BasicHttpEntity implements HttpAsyncCont
if (!queryStatus.hasError()) {
if (channel == null) {
- channel = Channels.newChannel(queryStatus.getResult(format));
+ String result = queryStatus.getResult(format);
+ log.info("Algorithm with queryId " + queryStatus.getQueryID().getQueryID()
+ + " terminated. Result: \n " + result);
+ channel = Channels.newChannel(
+ new ByteArrayInputStream(result.getBytes(StandardCharsets.UTF_8)));
}
channel.read(buffer);
buffer.flip();
@@ -68,32 +71,27 @@ public class NQueryResultEntity extends BasicHttpEntity implements HttpAsyncCont
close();
}
} else {
- log.trace("|" + queryStatus.getError() + "|");
if (queryStatus.getError().contains("ExaremeError:")) {
String data = queryStatus.getError().substring(queryStatus.getError().lastIndexOf("ExaremeError:") + "ExaremeError:".length()).replaceAll("\\s", " ");
- //type could be error, user_error, warning regarding the error occurred along the process
String result = HBPQueryHelper.ErrorResponse.createErrorResponse(data, user_error);
logErrorMessage(result);
encoder.write(ByteBuffer.wrap(result.getBytes()));
encoder.complete();
} else if (queryStatus.getError().contains("PrivacyError")) {
String data = "The Experiment could not run with the input provided because there are insufficient data.";
- //type could be error, user_error, warning regarding the error occurred along the process
String result = HBPQueryHelper.ErrorResponse.createErrorResponse(data, warning);
logErrorMessage(result);
encoder.write(ByteBuffer.wrap(result.getBytes()));
encoder.complete();
} else if (queryStatus.getError().contains("java.rmi.RemoteException")) {
String data = "One or more containers are not responding. Please inform the system administrator.";
- //type could be error, user_error, warning regarding the error occurred along the process
String result = HBPQueryHelper.ErrorResponse.createErrorResponse(data, error);
logErrorMessage(result);
encoder.write(ByteBuffer.wrap(result.getBytes()));
encoder.complete();
} else {
- log.info("Exception from madis: " + queryStatus.getError());
+ log.info("Exception when running the query: " + queryStatus.getError());
String data = "Something went wrong. Please inform the system administrator.";
- //type could be error, user_error, warning regarding the error occurred along the process
String result = HBPQueryHelper.ErrorResponse.createErrorResponse(data, error);
logErrorMessage(result);
encoder.write(ByteBuffer.wrap(result.getBytes()));
@@ -110,12 +108,11 @@ public class NQueryResultEntity extends BasicHttpEntity implements HttpAsyncCont
}
public void closeQuery() throws IOException {
- log.info("Closing from Query Result : " + queryStatus.getQueryID().getQueryID());
queryStatus.close();
}
@Override
- public void close() throws IOException {
+ public void close() {
}
diff --git a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/entity/NQueryStatusEntity.java b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/entity/NQueryStatusEntity.java
index 2986eccd61ccb5e05b9816172d814ade2c4f8e84..0db185cf98ab8e34f5343c9aaaf6466632227d3c 100644
--- a/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/entity/NQueryStatusEntity.java
+++ b/Exareme-Docker/src/exareme/exareme-master/src/main/java/madgik/exareme/master/gateway/async/handler/entity/NQueryStatusEntity.java
@@ -69,7 +69,7 @@ public class NQueryStatusEntity extends BasicHttpEntity implements HttpAsyncCont
}
@Override
- public void close() throws IOException {
+ public void close() {
}
diff --git a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/containerStatusMgr/PeriodicContainersStatusCheck.java b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/containerStatusMgr/PeriodicContainersStatusCheck.java
index 90f10e8316dec4d219eb0c6d5dac7002162f296d..2d1b1775e217a7fc95b715c89f67015f5dd74b22 100644
--- a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/containerStatusMgr/PeriodicContainersStatusCheck.java
+++ b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/containerStatusMgr/PeriodicContainersStatusCheck.java
@@ -31,7 +31,7 @@ public class PeriodicContainersStatusCheck {
}
- public void addConainerToCheck(EntityName container) {
+ public void addContainerToCheck(EntityName container) {
log.debug("Adding container to check: " + container);
containersToCheck.add(container);
}
diff --git a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/DynamicPlanManager.java b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/DynamicPlanManager.java
index aa73a8e3cc52c1bd6abf3754a7c90f420bb61918..fd2bf4c36f366b5a45f2a23ea46a4e725526ee26 100644
--- a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/DynamicPlanManager.java
+++ b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/DynamicPlanManager.java
@@ -25,6 +25,8 @@ import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
/**
* @author herald
@@ -34,6 +36,7 @@ public class DynamicPlanManager implements PlanSessionManagerInterface {
private final HashMap<PlanSessionID, PlanSessionReportID> elasticTreeSessions = new HashMap<>();
private EventProcessor eventProcessor = null;
private long sessionCount = 0;
+ ReentrantLock sessionCountLock = new ReentrantLock();
private long containerSessionCount = 0;
/* ROOT sessions */
private Map<PlanSessionID, PlanEventScheduler> schedulerMap = null;
@@ -71,11 +74,13 @@ public class DynamicPlanManager implements PlanSessionManagerInterface {
@Override
public void createGlobalScheduler() throws RemoteException {
+ sessionCountLock.lock();
PlanSessionID sessionID = new PlanSessionID(sessionCount);
PlanSessionReportID reportID = new PlanSessionReportID(sessionCount);
- reportID.reportManagerProxy = executionEngine.getPlanSessionReportManagerProxy(reportID);
sessionCount++;
+ sessionCountLock.unlock();
+ reportID.reportManagerProxy = executionEngine.getPlanSessionReportManagerProxy(reportID);
PlanEventScheduler eventScheduler =
new PlanEventScheduler(sessionID, reportID, eventProcessor, this, resourceManager,
registryProxy);
@@ -86,12 +91,13 @@ public class DynamicPlanManager implements PlanSessionManagerInterface {
@Override
public PlanSessionID createNewSession() throws RemoteException {
- // TODO Is this thread safe?
+ sessionCountLock.lock();
PlanSessionID sessionID = new PlanSessionID(sessionCount);
PlanSessionReportID reportID = new PlanSessionReportID(sessionCount);
- reportID.reportManagerProxy = executionEngine.getPlanSessionReportManagerProxy(reportID);
sessionCount++;
+ sessionCountLock.unlock();
+ reportID.reportManagerProxy = executionEngine.getPlanSessionReportManagerProxy(reportID);
PlanEventScheduler eventScheduler =
new PlanEventScheduler(sessionID, reportID, eventProcessor, this, resourceManager,
registryProxy);
@@ -105,8 +111,7 @@ public class DynamicPlanManager implements PlanSessionManagerInterface {
}
@Override
- public ContainerSessionID createContainerSession(PlanSessionID planSessionID)
- throws RemoteException {
+ public ContainerSessionID createContainerSession(PlanSessionID planSessionID) {
ContainerSessionID containerSessionID = new ContainerSessionID(containerSessionCount);
containerSessionCount++;
LinkedList<ContainerSessionID> containerSessionIDs = containerSessionMap.get(planSessionID);
diff --git a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/PlanEventScheduler.java b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/PlanEventScheduler.java
index f3b014da5af1ad3dadb4b4a69d4cb80ca0bed14b..1c2945a5025a9f13153f8b2f6e2963cd307a3544 100644
--- a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/PlanEventScheduler.java
+++ b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/PlanEventScheduler.java
@@ -301,7 +301,6 @@ public class PlanEventScheduler {
public void execute(ExecutionPlan plan) throws RemoteException {
lock.lock();
try {
- log.debug("Plan submitted for execution ...");
EditableExecutionPlan newPlan = preprocessPlan(plan);
validatePlan(newPlan);
log.debug("PlanAfterPreprocessing: " + newPlan.toString());
diff --git a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/PlanEventSchedulerState.java b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/PlanEventSchedulerState.java
index ebb0194890a4e210ee58f7cda1724d5653df2f71..69a70e32ff3df0537f8b333c66c1c5b888d5958b 100644
--- a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/PlanEventSchedulerState.java
+++ b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/PlanEventSchedulerState.java
@@ -308,7 +308,7 @@ public class PlanEventSchedulerState {
for (String containerName : plan.iterateContainers()) {
try {
if (!containerName.contains("any")) {
- pcsc.addConainerToCheck(plan.getContainer(containerName));
+ pcsc.addContainerToCheck(plan.getContainer(containerName));
}
} catch (SemanticError semanticError) {
semanticError.printStackTrace();
diff --git a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/event/planTermination/PlanTerminationEventHandler.java b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/event/planTermination/PlanTerminationEventHandler.java
index 364e011fb376acc74974acb521ab35a028d75bb7..98cc934ff893260e3630650c19d3db5d94e26b7d 100644
--- a/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/event/planTermination/PlanTerminationEventHandler.java
+++ b/Exareme-Docker/src/exareme/exareme-worker/src/main/java/madgik/exareme/worker/art/executionEngine/dynamicExecutionEngine/event/planTermination/PlanTerminationEventHandler.java
@@ -33,8 +33,8 @@ public class PlanTerminationEventHandler implements ExecEngineEventHandler<PlanT
try {
log.debug("closing session of container : " + proxy.getEntityName().getName());
proxy.destroySessions(state.getPlanSessionID());
- } catch (RemoteException e) {
- // state.addException(e);
+ } catch (Exception e) {
+ log.error("Cannot close the sessions for proxy: " + proxy, e);
// throw new ServerException("Cannot close all sessions", e);
}
}