You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by sh...@apache.org on 2016/01/06 20:17:13 UTC
[1/2] airavata git commit: Fixed AIRAVATA-1888
Repository: airavata
Updated Branches:
refs/heads/master 2fe5278f5 -> 98fc3f582
Fixed AIRAVATA-1888
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/5ce750d5
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/5ce750d5
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/5ce750d5
Branch: refs/heads/master
Commit: 5ce750d53bc4e8def467b9024c491a463e62203d
Parents: fc3f979
Author: Shameera Rathnayaka <sh...@gmail.com>
Authored: Wed Jan 6 14:15:35 2016 -0500
Committer: Shameera Rathnayaka <sh...@gmail.com>
Committed: Wed Jan 6 14:15:35 2016 -0500
----------------------------------------------------------------------
.../airavata/gfac/impl/GFacEngineImpl.java | 3 +-
.../apache/airavata/gfac/impl/GFacWorker.java | 32 +++-----------
.../airavata/gfac/server/GfacServerHandler.java | 44 +++++++++++++-------
3 files changed, 38 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/airavata/blob/5ce750d5/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java
index 9e6a522..dd10c12 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java
@@ -509,8 +509,9 @@ public class GFacEngineImpl implements GFacEngine {
processContext.setTaskExecutionOrder(taskExecutionOrder);
Map<String, TaskModel> taskMap = processContext.getTaskMap();
String recoverTaskId = null;
+ TaskModel taskModel = null;
for (String taskId : taskExecutionOrder) {
- TaskModel taskModel = taskMap.get(taskId);
+ taskModel = taskMap.get(taskId);
TaskState state = taskModel.getTaskStatus().getState();
if (state == TaskState.CREATED || state == TaskState.EXECUTING) {
recoverTaskId = taskId;
http://git-wip-us.apache.org/repos/asf/airavata/blob/5ce750d5/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacWorker.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacWorker.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacWorker.java
index c701ed5..e0664a5 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacWorker.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacWorker.java
@@ -78,14 +78,6 @@ public class GFacWorker implements Runnable {
@Override
public void run() {
try {
- if (processContext.isInterrupted()) {
- GFacUtils.handleProcessInterrupt(processContext);
- if (processContext.isCancel()) {
- sendAck();
- Factory.getGfacContext().removeProcess(processContext.getProcessId());
- }
- return;
- }
ProcessState processState = processContext.getProcessStatus().getState();
switch (processState) {
case CREATED:
@@ -206,7 +198,13 @@ public class GFacWorker implements Runnable {
}
private void executeProcess() throws GFacException {
+ // checkpoint
+ if (processContext.isInterrupted()) {
+ return;
+ }
+
engine.executeProcess(processContext);
+ // checkpoint
if (processContext.isInterrupted()) {
return;
}
@@ -216,24 +214,6 @@ public class GFacWorker implements Runnable {
}
}
-// private void monitorProcess() throws GFacException {
-// try {
-// JobMonitor monitorService = Factory.getMonitorService(processContext.getMonitorMode());
-// if (monitorService != null) {
-// monitorService.monitor(processContext.getJobModel().getJobId(), processContext);
-// ProcessStatus status = new ProcessStatus(ProcessState.MONITORING);
-// status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
-// processContext.setProcessStatus(status);
-// GFacUtils.saveAndPublishProcessStatus(processContext);
-// } else {
-// // we directly invoke outflow
-// continueTaskExecution();
-// }
-// } catch (AiravataException e) {
-// throw new GFacException("Error while retrieving moniot service", e);
-// }
-// }
-
private void sendAck() {
// this ensure, gfac doesn't send ack more than once for a process. which cause to remove gfac rabbitmq consumer from rabbitmq server.
if (!processContext.isAcknowledge()) {
http://git-wip-us.apache.org/repos/asf/airavata/blob/5ce750d5/modules/gfac/gfac-service/src/main/java/org/apache/airavata/gfac/server/GfacServerHandler.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-service/src/main/java/org/apache/airavata/gfac/server/GfacServerHandler.java b/modules/gfac/gfac-service/src/main/java/org/apache/airavata/gfac/server/GfacServerHandler.java
index c1c08a5..10da052 100644
--- a/modules/gfac/gfac-service/src/main/java/org/apache/airavata/gfac/server/GfacServerHandler.java
+++ b/modules/gfac/gfac-service/src/main/java/org/apache/airavata/gfac/server/GfacServerHandler.java
@@ -28,10 +28,8 @@ import org.apache.airavata.common.utils.ServerSettings;
import org.apache.airavata.common.utils.ThriftUtils;
import org.apache.airavata.common.utils.ZkConstants;
import org.apache.airavata.common.utils.listener.AbstractActivityListener;
-import org.apache.airavata.gfac.core.GFacConstants;
import org.apache.airavata.gfac.core.GFacException;
import org.apache.airavata.gfac.core.GFacUtils;
-import org.apache.airavata.gfac.core.watcher.CancelRequestWatcher;
import org.apache.airavata.gfac.cpi.GfacService;
import org.apache.airavata.gfac.cpi.gfac_cpiConstants;
import org.apache.airavata.gfac.impl.Factory;
@@ -250,7 +248,13 @@ public class GfacServerHandler implements GfacService.Iface {
publishProcessStatus(event, status);
try {
createProcessZKNode(curatorClient, gfacServerName, event, message);
- submitProcess(event.getProcessId(), event.getGatewayId(), event.getTokenId());
+ boolean isCancel = setCancelWatcher(curatorClient, event.getExperimentId(), event.getProcessId());
+ submitProcess(event.getProcessId(), event.getGatewayId(), event.getTokenId());
+ if (isCancel) {
+ // Need to trigger process cancel watcher, wait till process recover and then set zk data.
+ Thread.sleep(10000);
+ setCancelData(event.getExperimentId());
+ }
} catch (Exception e) {
log.error(e.getMessage(), e);
rabbitMQProcessLaunchConsumer.sendAck(message.getDeliveryTag());
@@ -262,7 +266,9 @@ public class GfacServerHandler implements GfacService.Iface {
} catch (AiravataException e) {
log.error("Error while publishing process status", e);
}
- } else if (message.getType().equals(MessageType.TERMINATEPROCESS)) {
+ }
+ // TODO - Now there is no process termination type messages, use zookeeper instead of rabbitmq to do that. it is safe to remove this else part.
+ else if (message.getType().equals(MessageType.TERMINATEPROCESS)) {
ProcessTerminateEvent event = new ProcessTerminateEvent();
TBase messageEvent = message.getEvent();
try {
@@ -289,7 +295,26 @@ public class GfacServerHandler implements GfacService.Iface {
}
}
- private void publishProcessStatus(ProcessSubmitEvent event, ProcessStatus status) throws AiravataException {
+ private void setCancelData(String experimentId) throws Exception {
+ String expCancelNodePath = ZKPaths.makePath(ZKPaths.makePath(ZkConstants.ZOOKEEPER_EXPERIMENT_NODE,
+ experimentId), ZkConstants.ZOOKEEPER_CANCEL_LISTENER_NODE);
+ curatorClient.setData().withVersion(-1).forPath(expCancelNodePath, ZkConstants.ZOOKEEPER_CANCEL_REQEUST
+ .getBytes());
+ }
+
+ private boolean setCancelWatcher(CuratorFramework curatorClient,
+ String experimentId,
+ String processId) throws Exception {
+
+ String experimentNodePath = GFacUtils.getExperimentNodePath(experimentId);
+ // create /experiments/{experimentId}/cancel node and set watcher for data changes
+ String experimentCancelNode = ZKPaths.makePath(experimentNodePath, ZkConstants.ZOOKEEPER_CANCEL_LISTENER_NODE);
+ byte[] bytes = curatorClient.getData().usingWatcher(Factory.getCancelRequestWatcher(experimentId, processId)).forPath(experimentCancelNode);
+ return bytes != null && new String(bytes).equalsIgnoreCase(ZkConstants.ZOOKEEPER_CANCEL_REQEUST);
+
+ }
+
+ private void publishProcessStatus(ProcessSubmitEvent event, ProcessStatus status) throws AiravataException {
ProcessIdentifier identifier = new ProcessIdentifier(event.getProcessId(),
event.getExperimentId(),
event.getGatewayId());
@@ -324,15 +349,6 @@ public class GfacServerHandler implements GfacService.Iface {
ZKPaths.mkdirs(curatorClient.getZookeeperClient().getZooKeeper(), tokenNodePath);
curatorClient.setData().withVersion(-1).forPath(tokenNodePath, token.getBytes());
- // create /experiments/{experimentId}/{processId}/cancelListener node and set watcher for data changes
-/* String cancelListenerNode = ZKPaths.makePath(zkProcessNodePath, GFacConstants.ZOOKEEPER_CANCEL_LISTENER_NODE);
- ZKPaths.mkdirs(curatorClient.getZookeeperClient().getZooKeeper(), cancelListenerNode);
- curatorClient.getData().usingWatcher(Factory.getCancelRequestWatcher()).forPath(cancelListenerNode);*/
-
- // create /experiments/{experimentId}/cancel node and set watcher for data changes
- String experimentCancelNode = ZKPaths.makePath(experimentNodePath, ZkConstants.ZOOKEEPER_CANCEL_LISTENER_NODE);
- curatorClient.getData().usingWatcher(Factory.getCancelRequestWatcher(experimentId, processId)).forPath (experimentCancelNode);
-
}
private void updateDeliveryTag(CuratorFramework curatorClient, String gfacServerName, ProcessSubmitEvent event,
[2/2] airavata git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/airavata
Posted by sh...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/airavata
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/98fc3f58
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/98fc3f58
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/98fc3f58
Branch: refs/heads/master
Commit: 98fc3f58280d126f2c6fd1d03477f8b9a9332838
Parents: 5ce750d 2fe5278
Author: Shameera Rathnayaka <sh...@gmail.com>
Authored: Wed Jan 6 14:15:50 2016 -0500
Committer: Shameera Rathnayaka <sh...@gmail.com>
Committed: Wed Jan 6 14:15:50 2016 -0500
----------------------------------------------------------------------
.../server/handler/AiravataServerHandler.java | 166 ++++++++++++++-----
.../java/org/apache/airavata/api/Airavata.java | 2 +-
.../lib/airavata/experiment_model_types.cpp | 22 +++
.../lib/airavata/experiment_model_types.h | 12 +-
.../lib/airavata/process_model_types.cpp | 22 +++
.../lib/airavata/process_model_types.h | 12 +-
.../lib/Airavata/Model/Experiment/Types.php | 23 +++
.../lib/Airavata/Model/Process/Types.php | 23 +++
.../apache/airavata/model/experiment/ttypes.py | 15 +-
.../lib/apache/airavata/model/process/ttypes.py | 15 +-
.../org/apache/airavata/model/Workflow.java | 2 +-
.../ApplicationDeploymentDescription.java | 2 +-
.../appdeployment/ApplicationModule.java | 2 +-
.../appcatalog/appdeployment/CommandObject.java | 2 +-
.../appcatalog/appdeployment/SetEnvPaths.java | 2 +-
.../ApplicationInterfaceDescription.java | 2 +-
.../appcatalog/computeresource/BatchQueue.java | 2 +-
.../computeresource/CloudJobSubmission.java | 2 +-
.../ComputeResourceDescription.java | 2 +-
.../computeresource/GlobusJobSubmission.java | 2 +-
.../computeresource/JobSubmissionInterface.java | 2 +-
.../computeresource/LOCALSubmission.java | 2 +-
.../computeresource/ResourceJobManager.java | 2 +-
.../computeresource/SSHJobSubmission.java | 2 +-
.../computeresource/UnicoreJobSubmission.java | 2 +-
.../ComputeResourcePreference.java | 2 +-
.../gatewayprofile/GatewayResourceProfile.java | 2 +-
.../gatewayprofile/StoragePreference.java | 2 +-
.../StorageResourceDescription.java | 2 +-
.../application/io/InputDataObjectType.java | 2 +-
.../application/io/OutputDataObjectType.java | 2 +-
.../airavata/model/commons/ErrorModel.java | 2 +-
.../model/commons/ValidationResults.java | 2 +-
.../airavata/model/commons/ValidatorResult.java | 2 +-
.../data/movement/DataMovementInterface.java | 2 +-
.../data/movement/GridFTPDataMovement.java | 2 +-
.../model/data/movement/LOCALDataMovement.java | 2 +-
.../model/data/movement/SCPDataMovement.java | 2 +-
.../data/movement/UnicoreDataMovement.java | 2 +-
.../data/resource/DataReplicaLocationModel.java | 2 +-
.../model/data/resource/DataResourceModel.java | 2 +-
.../model/error/AiravataClientException.java | 2 +-
.../model/error/AiravataSystemException.java | 2 +-
.../model/error/AuthenticationException.java | 2 +-
.../model/error/AuthorizationException.java | 2 +-
.../error/ExperimentNotFoundException.java | 2 +-
.../model/error/InvalidRequestException.java | 2 +-
.../model/error/LaunchValidationException.java | 2 +-
.../model/error/ProjectNotFoundException.java | 2 +-
.../airavata/model/error/TimedOutException.java | 2 +-
.../airavata/model/error/ValidationResults.java | 2 +-
.../airavata/model/error/ValidatorResult.java | 2 +-
.../model/experiment/ExperimentModel.java | 2 +-
.../model/experiment/ExperimentStatistics.java | 2 +-
.../experiment/ExperimentSummaryModel.java | 2 +-
.../experiment/UserConfigurationDataModel.java | 116 ++++++++++++-
.../org/apache/airavata/model/job/JobModel.java | 2 +-
.../event/ExperimentStatusChangeEvent.java | 2 +-
.../model/messaging/event/JobIdentifier.java | 2 +-
.../messaging/event/JobStatusChangeEvent.java | 2 +-
.../event/JobStatusChangeRequestEvent.java | 2 +-
.../airavata/model/messaging/event/Message.java | 2 +-
.../messaging/event/ProcessIdentifier.java | 2 +-
.../event/ProcessStatusChangeEvent.java | 2 +-
.../event/ProcessStatusChangeRequestEvent.java | 2 +-
.../messaging/event/ProcessSubmitEvent.java | 2 +-
.../messaging/event/ProcessTerminateEvent.java | 2 +-
.../model/messaging/event/TaskIdentifier.java | 2 +-
.../messaging/event/TaskOutputChangeEvent.java | 2 +-
.../messaging/event/TaskStatusChangeEvent.java | 2 +-
.../event/TaskStatusChangeRequestEvent.java | 2 +-
.../airavata/model/process/ProcessModel.java | 116 ++++++++++++-
.../ComputationalResourceSchedulingModel.java | 2 +-
.../airavata/model/security/AuthzToken.java | 2 +-
.../airavata/model/status/ExperimentStatus.java | 2 +-
.../apache/airavata/model/status/JobStatus.java | 2 +-
.../airavata/model/status/ProcessStatus.java | 2 +-
.../airavata/model/status/TaskStatus.java | 2 +-
.../model/task/DataStagingTaskModel.java | 2 +-
.../model/task/EnvironmentSetupTaskModel.java | 2 +-
.../model/task/JobSubmissionTaskModel.java | 2 +-
.../airavata/model/task/MonitorTaskModel.java | 2 +-
.../apache/airavata/model/task/TaskModel.java | 2 +-
.../airavata/model/workspace/Gateway.java | 2 +-
.../apache/airavata/model/workspace/Group.java | 2 +-
.../airavata/model/workspace/Project.java | 2 +-
.../apache/airavata/model/workspace/User.java | 2 +-
.../model/util/ExperimentModelUtil.java | 1 +
.../gfac/impl/task/SCPDataStageTask.java | 17 +-
.../catalog/impl/ExperimentRegistry.java | 4 +
.../core/experiment/catalog/model/Process.java | 10 ++
.../catalog/model/UserConfigurationData.java | 11 ++
.../catalog/resources/ProcessResource.java | 10 ++
.../UserConfigurationDataResource.java | 10 ++
.../experiment/catalog/resources/Utils.java | 1 +
.../utils/ThriftDataModelConversion.java | 2 +
.../src/main/resources/expcatalog-derby.sql | 2 +
.../src/main/resources/expcatalog-mysql.sql | 2 +
.../experiment_model.thrift | 1 +
.../process_model.thrift | 3 +-
100 files changed, 627 insertions(+), 141 deletions(-)
----------------------------------------------------------------------