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 2019/12/11 16:17:55 UTC
[airavata-mft] branch master updated: Async transfer monitoring
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
The following commit(s) were added to refs/heads/master by this push:
new 84f60ac Async transfer monitoring
84f60ac is described below
commit 84f60ac61b2c406df7b5985ea9445afc0f8f21b4
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Wed Dec 11 11:17:41 2019 -0500
Async transfer monitoring
---
.../org/apache/airavata/mft/agent/MFTAgent.java | 6 +--
.../airavata/mft/agent/TransportMediator.java | 63 ++++++++++++++++------
2 files changed, 49 insertions(+), 20 deletions(-)
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java b/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java
index 1b050f0..d788995 100644
--- a/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java
@@ -30,8 +30,8 @@ public class MFTAgent {
MetadataCollector metadataCollector = resolveMetadataCollector(request.getSourceType());
ResourceMetadata metadata = metadataCollector.getGetResourceMetadata(request.getSourceId(), request.getSourceToken());
System.out.println("File size " + metadata.getResourceSize());
- mediator.transfer(inConnector, outConnector, metadata);
-
+ String transferId = mediator.transfer(inConnector, outConnector, metadata);
+ System.out.println("Submitted transfer " + transferId);
} catch (Exception e) {
e.printStackTrace();
}
@@ -42,7 +42,7 @@ public class MFTAgent {
request.setSourceId("1");
request.setSourceType("SCP");
request.setDestinationId("2");
- request.setDestinationType("LOCAL");
+ request.setDestinationType("SCP");
MFTAgent agent = new MFTAgent();
agent.requests.add(request);
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java b/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java
index f20570d..dec9cdd 100644
--- a/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java
@@ -8,17 +8,21 @@ import org.apache.airavata.mft.core.api.Connector;
import java.util.ArrayList;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.*;
public class TransportMediator {
private ExecutorService executor = Executors.newFixedThreadPool(10);
+ private ExecutorService monitor = Executors.newFixedThreadPool(10);
public void destroy() {
executor.shutdown();
}
- public void transfer(Connector inConnector, Connector outConnector, ResourceMetadata metadata) throws Exception {
+ public String transfer(Connector inConnector, Connector outConnector, ResourceMetadata metadata) throws Exception {
+
+ String transferId = UUID.randomUUID().toString();
CircularStreamingBuffer streamBuffer = new CircularStreamingBuffer();
ConnectorContext context = new ConnectorContext();
@@ -31,28 +35,53 @@ public class TransportMediator {
ExecutorCompletionService<Integer> completionService = new ExecutorCompletionService<>(executor);
+ long startTime = System.currentTimeMillis();
+
futureList.add(completionService.submit(recvTask));
futureList.add(completionService.submit(sendTask));
- for (int i = 0; i < futureList.size(); i++) {
- Future<Integer> ft = completionService.take();
- futureList.remove(ft);
- try {
- ft.get();
- } catch(InterruptedException e){
- // Interrupted
- } catch(ExecutionException e){
- // Snap, something went wrong in the task! Abort! Abort! Abort!
- System.out.println("One task failed with error: " + e.getMessage() );
- e.printStackTrace();
- for(Future<Integer> f : futureList){
- Thread.sleep(1000);
- f.cancel(true);
+ Thread monitorThread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+
+ try {
+ for (int i = 0; i < futureList.size(); i++) {
+ Future<Integer> ft = completionService.take();
+ futureList.remove(ft);
+ try {
+ ft.get();
+ } catch (InterruptedException e) {
+ // Interrupted
+ } catch (ExecutionException e) {
+ // Snap, something went wrong in the task! Abort! Abort! Abort!
+ System.out.println("One task failed with error: " + e.getMessage());
+ e.printStackTrace();
+ for (Future<Integer> f : futureList) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException ex) {
+ ex.printStackTrace();
+ }
+ f.cancel(true);
+ }
+ futureList.clear();
+ }
+ }
+
+ long endTime = System.currentTimeMillis();
+
+ long time = (endTime - startTime) / 1000;
+
+ System.out.println("Transfer Speed " + (metadata.getResourceSize() * 1.0 / time) / (1024 * 1024) + " MB/s");
+ System.out.println("Transfer " + transferId + " completed");
+ } catch (Exception e) {
+ e.printStackTrace();
}
- futureList.clear();
}
- }
+ });
+ monitor.submit(monitorThread);
+ return transferId;
//inConnector.destroy();
//outConnector.destroy();
}