You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by di...@apache.org on 2021/12/15 21:14:32 UTC
[airavata] branch AIRAVATA-3549 updated: Supporting intermediate transfers at helix level
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch AIRAVATA-3549
in repository https://gitbox.apache.org/repos/asf/airavata.git
The following commit(s) were added to refs/heads/AIRAVATA-3549 by this push:
new cef65af Supporting intermediate transfers at helix level
cef65af is described below
commit cef65af8277a8ab3a9087a4083f16f2d977ac639
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Wed Dec 15 16:14:22 2021 -0500
Supporting intermediate transfers at helix level
---
.../apache/airavata/helix/impl/task/AiravataTask.java | 11 ++++++++---
.../helix/impl/task/staging/OutputDataStagingTask.java | 7 +++++--
.../helix/impl/workflow/PreWorkflowManager.java | 17 ++++++++++++++++-
3 files changed, 29 insertions(+), 6 deletions(-)
diff --git a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/AiravataTask.java b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/AiravataTask.java
index 1379307..c6cf095 100644
--- a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/AiravataTask.java
+++ b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/AiravataTask.java
@@ -259,7 +259,7 @@ public abstract class AiravataTask extends AbstractTask {
}
}
- public void saveExperimentOutput(String outputName, String outputVal) throws TaskOnFailException {
+ public void saveExperimentOutput(String outputName, String outputVal, boolean onlyUpdateProcess) throws TaskOnFailException {
try {
ExperimentModel experiment = getRegistryServiceClient().getExperiment(experimentId);
List<OutputDataObjectType> experimentOutputs = experiment.getExperimentOutputs();
@@ -268,8 +268,13 @@ public abstract class AiravataTask extends AbstractTask {
if (expOutput.getName().equals(outputName)) {
String productUri = saveDataProduct(outputName, outputVal, expOutput.getMetaData());
expOutput.setValue(productUri);
- getRegistryServiceClient().addExperimentProcessOutputs("EXPERIMENT_OUTPUT",
- Collections.singletonList(expOutput), experimentId);
+
+ if (!onlyUpdateProcess) {
+ getRegistryServiceClient().addExperimentProcessOutputs("EXPERIMENT_OUTPUT",
+ Collections.singletonList(expOutput), experimentId);
+ }
+ getRegistryServiceClient().addExperimentProcessOutputs("PROCESS_OUTPUT",
+ Collections.singletonList(expOutput), processId);
}
}
}
diff --git a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/staging/OutputDataStagingTask.java b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/staging/OutputDataStagingTask.java
index 4419250..dbdd5f3 100644
--- a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/staging/OutputDataStagingTask.java
+++ b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/staging/OutputDataStagingTask.java
@@ -30,7 +30,9 @@ import org.apache.airavata.model.appcatalog.storageresource.StorageResourceDescr
import org.apache.airavata.model.application.io.DataType;
import org.apache.airavata.model.application.io.OutputDataObjectType;
import org.apache.airavata.model.status.ProcessState;
+import org.apache.airavata.model.task.DataStageType;
import org.apache.airavata.model.task.DataStagingTaskModel;
+import org.apache.airavata.model.task.TaskTypes;
import org.apache.airavata.patform.monitoring.CountMonitor;
import org.apache.helix.task.TaskResult;
import org.apache.thrift.TException;
@@ -61,6 +63,7 @@ public class OutputDataStagingTask extends DataStagingTask {
try {
// Get and validate data staging task model
DataStagingTaskModel dataStagingTaskModel = getDataStagingTaskModel();
+ boolean onlyUpdateProcess = taskContext.getCurrentTaskModel().getTaskType() == TaskTypes.OUTPUT_FETCHING;
// Fetch and validate input data type from data staging task model
OutputDataObjectType processOutput = dataStagingTaskModel.getProcessOutput();
@@ -169,7 +172,7 @@ public class OutputDataStagingTask extends DataStagingTask {
}
if (!destinationURIs.isEmpty()) {
if (processOutput.getType() == DataType.URI) {
- saveExperimentOutput(processOutput.getName(), destinationURIs.get(0).toString());
+ saveExperimentOutput(processOutput.getName(), destinationURIs.get(0).toString(), onlyUpdateProcess);
} else if (processOutput.getType() == DataType.URI_COLLECTION) {
saveExperimentOutputCollection(processOutput.getName(), destinationURIs.stream().map(URI::toString).collect(Collectors.toList()));
}
@@ -181,7 +184,7 @@ public class OutputDataStagingTask extends DataStagingTask {
assert processOutput != null;
boolean transferred = transferFileToStorage(sourceURI.getPath(), destinationURI.getPath(), sourceFileName, adaptor, storageResourceAdaptor);
if (transferred) {
- saveExperimentOutput(processOutput.getName(), destinationURI.toString());
+ saveExperimentOutput(processOutput.getName(), destinationURI.toString(), onlyUpdateProcess);
} else {
logger.warn("File " + sourceFileName + " did not transfer");
}
diff --git a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PreWorkflowManager.java b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PreWorkflowManager.java
index 3bb4c81..0018be7 100644
--- a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PreWorkflowManager.java
+++ b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PreWorkflowManager.java
@@ -31,6 +31,7 @@ import org.apache.airavata.helix.impl.task.cancel.RemoteJobCancellationTask;
import org.apache.airavata.helix.impl.task.cancel.WorkflowCancellationTask;
import org.apache.airavata.helix.impl.task.env.EnvSetupTask;
import org.apache.airavata.helix.impl.task.staging.InputDataStagingTask;
+import org.apache.airavata.helix.impl.task.staging.OutputDataStagingTask;
import org.apache.airavata.helix.impl.task.submission.DefaultJobSubmissionTask;
import org.apache.airavata.messaging.core.*;
import org.apache.airavata.model.experiment.ExperimentModel;
@@ -100,6 +101,13 @@ public class PreWorkflowManager extends WorkflowManager {
String taskDag = processModel.getTaskDag();
List<TaskModel> taskList = processModel.getTasks();
+ boolean intermediateTransfer = taskList.stream()
+ .anyMatch(task -> task.getTaskType() == TaskTypes.OUTPUT_FETCHING);
+
+ if (intermediateTransfer) {
+ logger.info("Process {} contains intermediate file transfers", processId);
+ }
+
String[] taskIds = taskDag.split(",");
final List<AiravataTask> allTasks = new ArrayList<>();
@@ -111,7 +119,14 @@ public class PreWorkflowManager extends WorkflowManager {
if (model.isPresent()) {
TaskModel taskModel = model.get();
AiravataTask airavataTask = null;
- if (taskModel.getTaskType() == TaskTypes.ENV_SETUP) {
+
+ if (intermediateTransfer) {
+ if (taskModel.getTaskType() == TaskTypes.OUTPUT_FETCHING) {
+ airavataTask = new OutputDataStagingTask();
+ airavataTask.setForceRunTask(true);
+ }
+
+ } else if (taskModel.getTaskType() == TaskTypes.ENV_SETUP) {
airavataTask = new EnvSetupTask();
airavataTask.setForceRunTask(true);
} else if (taskModel.getTaskType() == TaskTypes.JOB_SUBMISSION) {