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 2022/04/05 23:22:52 UTC
[airavata-mft] branch master updated: Decoupling transfer request accept and process threads in Agent for optimal performance
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 2b2eab6 Decoupling transfer request accept and process threads in Agent for optimal performance
2b2eab6 is described below
commit 2b2eab69ce4888b7fe3ab2ded844eab9f8384671
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Tue Apr 5 19:22:43 2022 -0400
Decoupling transfer request accept and process threads in Agent for optimal performance
---
.../org/apache/airavata/mft/agent/MFTAgent.java | 218 +++++++++---------
.../airavata/mft/agent/TransportMediator.java | 247 ++++++++++-----------
.../mft/core/api/IncomingChunkedConnector.java | 2 +
.../mft/core/api/OutgoingChunkedConnector.java | 2 +
.../mft/transport/s3/S3IncomingConnector.java | 4 +-
.../mft/transport/s3/S3OutgoingConnector.java | 3 +-
6 files changed, 241 insertions(+), 235 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 15f316e..5a61d31 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
@@ -55,10 +55,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
public class MFTAgent implements CommandLineRunner {
@@ -122,6 +119,8 @@ public class MFTAgent implements CommandLineRunner {
private long sessionRenewSeconds = 4;
private long sessionTTLSeconds = 10;
private String session;
+ private ExecutorService transferRequestExecutor;
+
private TransportMediator mediator;
@@ -141,6 +140,7 @@ public class MFTAgent implements CommandLineRunner {
transferMessageCache = KVCache.newCache(mftConsulClient.getKvClient(), MFTConsulClient.AGENTS_TRANSFER_REQUEST_MESSAGE_PATH + agentId);
rpcMessageCache = KVCache.newCache(mftConsulClient.getKvClient(), MFTConsulClient.AGENTS_RPC_REQUEST_MESSAGE_PATH + agentId);
mediator = new TransportMediator(tempDataDir, concurrentTransfers, concurrentChunkedThreads, chunkedSize);
+ transferRequestExecutor = Executors.newFixedThreadPool(concurrentTransfers);
}
private void acceptRPCRequests() {
@@ -164,6 +164,110 @@ public class MFTAgent implements CommandLineRunner {
rpcMessageCache.start();
}
+ private void processTransfer(String transferId, String transferRequestJson) {
+ logger.info("Received raw message: {}", transferRequestJson);
+ TransferApiRequest request = null;
+ try {
+ TransferApiRequest.Builder builder = TransferApiRequest.newBuilder();
+ JsonFormat.parser().merge(transferRequestJson, builder);
+ request = builder.build();
+
+ logger.info("Received request " + transferId);
+ mftConsulClient.submitTransferStateToProcess(transferId, agentId, new TransferState()
+ .setState("STARTING")
+ .setPercentage(0)
+ .setUpdateTimeMils(System.currentTimeMillis())
+ .setPublisher(agentId)
+ .setDescription("Starting the transfer"));
+
+ Optional<MetadataCollector> srcMetadataCollectorOp = MetadataCollectorResolver.resolveMetadataCollector(request.getSourceType());
+ MetadataCollector srcMetadataCollector = srcMetadataCollectorOp.orElseThrow(() -> new Exception("Could not find a metadata collector for source"));
+ srcMetadataCollector.init(resourceServiceHost, resourceServicePort, secretServiceHost, secretServicePort);
+
+ Optional<MetadataCollector> dstMetadataCollectorOp = MetadataCollectorResolver.resolveMetadataCollector(request.getDestinationType());
+ MetadataCollector dstMetadataCollector = dstMetadataCollectorOp.orElseThrow(() -> new Exception("Could not find a metadata collector for destination"));
+ dstMetadataCollector.init(resourceServiceHost, resourceServicePort, secretServiceHost, secretServicePort);
+
+ FileResourceMetadata srcMetadata = srcMetadataCollector.getFileResourceMetadata(
+ request.getMftAuthorizationToken(),
+ request.getSourceResourceId(),
+ request.getSourceToken());
+
+
+ ConnectorConfig srcCC = ConnectorConfig.ConnectorConfigBuilder.newBuilder()
+ .withAuthToken(request.getMftAuthorizationToken())
+ .withResourceServiceHost(resourceServiceHost)
+ .withResourceServicePort(resourceServicePort)
+ .withSecretServiceHost(secretServiceHost)
+ .withSecretServicePort(secretServicePort)
+ .withTransferId(transferId)
+ .withResourceId(request.getSourceResourceId())
+ .withCredentialToken(request.getSourceToken())
+ .withMetadata(srcMetadata).build();
+
+ ConnectorConfig dstCC = ConnectorConfig.ConnectorConfigBuilder.newBuilder()
+ .withAuthToken(request.getMftAuthorizationToken())
+ .withResourceServiceHost(resourceServiceHost)
+ .withResourceServicePort(resourceServicePort)
+ .withSecretServiceHost(secretServiceHost)
+ .withSecretServicePort(secretServicePort)
+ .withTransferId(transferId)
+ .withResourceId(request.getDestinationResourceId())
+ .withCredentialToken(request.getDestinationToken())
+ .withMetadata(srcMetadata).build();
+
+ mftConsulClient.submitTransferStateToProcess(transferId, agentId, new TransferState()
+ .setState("STARTED")
+ .setPercentage(0)
+ .setUpdateTimeMils(System.currentTimeMillis())
+ .setPublisher(agentId)
+ .setDescription("Started the transfer"));
+
+ // Save transfer metadata in scheduled path to recover in case of an Agent failures. Recovery is done from controller
+ mftConsulClient.getKvClient().putValue(MFTConsulClient.AGENTS_SCHEDULED_PATH + agentId + "/" + session + "/" + transferId, transferRequestJson);
+
+ mediator.transferSingleThread(transferId, request, srcCC, dstCC,
+ (id, st) -> {
+ try {
+ mftConsulClient.submitTransferStateToProcess(id, agentId, st.setPublisher(agentId));
+
+ } catch (MFTConsulClientException e) {
+ logger.error("Failed while updating transfer state", e);
+ }
+ },
+ (id, transferSuccess) -> {
+ try {
+ // Delete scheduled key as the transfer completed / failed if it was placed in current session
+ mftConsulClient.getKvClient().deleteKey(MFTConsulClient.AGENTS_SCHEDULED_PATH + agentId + "/" + session + "/" + id);
+ } catch (Exception e) {
+ logger.error("Failed while deleting scheduled path for transfer {}", id);
+ }
+ });
+
+
+ } catch (Throwable e) {
+ if (request != null) {
+ try {
+ logger.error("Error in submitting transfer {}", transferId, e);
+
+ mftConsulClient.submitTransferStateToProcess(transferId, agentId, new TransferState()
+ .setState("FAILED")
+ .setPercentage(0)
+ .setUpdateTimeMils(System.currentTimeMillis())
+ .setPublisher(agentId)
+ .setDescription(ExceptionUtils.getStackTrace(e)));
+ } catch (MFTConsulClientException ex) {
+ logger.warn(ex.getMessage());
+ // Ignore
+ }
+ } else {
+ logger.error("Unknown error in processing message {}", transferRequestJson, e);
+ }
+ } finally {
+ //logger.info("Deleting key " + consulEntryKey);
+ //mftConsulClient.getKvClient().deleteKey(consulEntryKey); // Due to bug in consul https://github.com/hashicorp/consul/issues/571
+ }
+ }
private void acceptTransferRequests() {
transferCacheListener = newValues -> {
@@ -171,110 +275,9 @@ public class MFTAgent implements CommandLineRunner {
Optional<String> decodedValue = value.getValueAsString();
String transferId = value.getKey().substring(value.getKey().lastIndexOf("/") + 1);
decodedValue.ifPresent(v -> {
- logger.info("Received raw message: {}", v);
- TransferApiRequest request = null;
- try {
- TransferApiRequest.Builder builder = TransferApiRequest.newBuilder();
- JsonFormat.parser().merge(v, builder);
- request = builder.build();
-
- logger.info("Received request " + transferId);
- mftConsulClient.submitTransferStateToProcess(transferId, agentId, new TransferState()
- .setState("STARTING")
- .setPercentage(0)
- .setUpdateTimeMils(System.currentTimeMillis())
- .setPublisher(agentId)
- .setDescription("Starting the transfer"));
-
- Optional<MetadataCollector> srcMetadataCollectorOp = MetadataCollectorResolver.resolveMetadataCollector(request.getSourceType());
- MetadataCollector srcMetadataCollector = srcMetadataCollectorOp.orElseThrow(() -> new Exception("Could not find a metadata collector for source"));
- srcMetadataCollector.init(resourceServiceHost, resourceServicePort, secretServiceHost, secretServicePort);
-
- Optional<MetadataCollector> dstMetadataCollectorOp = MetadataCollectorResolver.resolveMetadataCollector(request.getDestinationType());
- MetadataCollector dstMetadataCollector = dstMetadataCollectorOp.orElseThrow(() -> new Exception("Could not find a metadata collector for destination"));
- dstMetadataCollector.init(resourceServiceHost, resourceServicePort, secretServiceHost, secretServicePort);
-
- FileResourceMetadata srcMetadata = srcMetadataCollector.getFileResourceMetadata(
- request.getMftAuthorizationToken(),
- request.getSourceResourceId(),
- request.getSourceToken());
-
-
- ConnectorConfig srcCC = ConnectorConfig.ConnectorConfigBuilder.newBuilder()
- .withAuthToken(request.getMftAuthorizationToken())
- .withResourceServiceHost(resourceServiceHost)
- .withResourceServicePort(resourceServicePort)
- .withSecretServiceHost(secretServiceHost)
- .withSecretServicePort(secretServicePort)
- .withTransferId(transferId)
- .withResourceId(request.getSourceResourceId())
- .withCredentialToken(request.getSourceToken())
- .withMetadata(srcMetadata).build();
-
- ConnectorConfig dstCC = ConnectorConfig.ConnectorConfigBuilder.newBuilder()
- .withAuthToken(request.getMftAuthorizationToken())
- .withResourceServiceHost(resourceServiceHost)
- .withResourceServicePort(resourceServicePort)
- .withSecretServiceHost(secretServiceHost)
- .withSecretServicePort(secretServicePort)
- .withTransferId(transferId)
- .withResourceId(request.getDestinationResourceId())
- .withCredentialToken(request.getDestinationToken())
- .withMetadata(srcMetadata).build();
-
- mftConsulClient.submitTransferStateToProcess(transferId, agentId, new TransferState()
- .setState("STARTED")
- .setPercentage(0)
- .setUpdateTimeMils(System.currentTimeMillis())
- .setPublisher(agentId)
- .setDescription("Started the transfer"));
-
- mediator.transferSingleThread(transferId, request, srcCC, dstCC,
- (id, st) -> {
- try {
- mftConsulClient.submitTransferStateToProcess(id, agentId, st.setPublisher(agentId));
-
- } catch (MFTConsulClientException e) {
- logger.error("Failed while updating transfer state", e);
- }
- },
- (id, transferSuccess) -> {
- try {
- // Delete scheduled key as the transfer completed / failed if it was placed in current session
- mftConsulClient.getKvClient().deleteKey(MFTConsulClient.AGENTS_SCHEDULED_PATH + agentId + "/" + session + "/" + id);
- } catch (Exception e) {
- logger.error("Failed while deleting scheduled path for transfer {}", id);
- }
- });
-
- logger.info("Started the transfer " + transferId);
-
- // Save transfer metadata in scheduled path to recover in case of an Agent failures. Recovery is done from controller
- mftConsulClient.getKvClient().putValue(MFTConsulClient.AGENTS_SCHEDULED_PATH + agentId + "/" + session + "/" + transferId, v);
- } catch (Throwable e) {
- if (request != null) {
- try {
- logger.error("Error in submitting transfer {}", transferId, e);
-
- mftConsulClient.submitTransferStateToProcess(transferId, agentId, new TransferState()
- .setState("FAILED")
- .setPercentage(0)
- .setUpdateTimeMils(System.currentTimeMillis())
- .setPublisher(agentId)
- .setDescription(ExceptionUtils.getStackTrace(e)));
- } catch (MFTConsulClientException ex) {
- logger.warn(ex.getMessage());
- // Ignore
- }
- } else {
- logger.error("Unknown error in processing message {}", v, e);
- }
- } finally {
- logger.info("Deleting key " + value.getKey());
- mftConsulClient.getKvClient().deleteKey(value.getKey()); // Due to bug in consul https://github.com/hashicorp/consul/issues/571
- }
+ mftConsulClient.getKvClient().deleteKey(value.getKey());
+ transferRequestExecutor.submit(() -> processTransfer(transferId, v));
});
-
});
};
transferMessageCache.addListener(transferCacheListener);
@@ -380,6 +383,7 @@ public class MFTAgent implements CommandLineRunner {
logger.info("Stopping Agent " + agentId);
disconnectAgent();
mainHold.release();
+ transferRequestExecutor.shutdown();
}
public void start() throws Exception {
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 581f4b5..49de2f8 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
@@ -41,19 +41,18 @@ public class TransportMediator {
/*
Number of maximum transfers handled at atime
*/
- private final ExecutorService executor;
private final ExecutorService monitorPool;
private String tempDataDir = "/tmp";
- private final int concurrentChunkedThreads;
private final int chunkedSize;
+ private final ExecutorService chunkedExecutorService;
+
public TransportMediator(String tempDataDir, int concurrentTransfers, int concurrentChunkedThreads, int chunkedSize) {
this.tempDataDir = tempDataDir;
- executor = Executors.newFixedThreadPool(concurrentTransfers);
monitorPool = Executors.newFixedThreadPool(concurrentTransfers);
- this.concurrentChunkedThreads = concurrentChunkedThreads;
this.chunkedSize = chunkedSize;
+ chunkedExecutorService = Executors.newFixedThreadPool(concurrentChunkedThreads);
}
public void transferSingleThread(String transferId,
@@ -63,166 +62,162 @@ public class TransportMediator {
BiConsumer<String, TransferState> onStatusCallback,
BiConsumer<String, Boolean> exitingCallback) {
- executor.submit(() -> {
+ final AtomicBoolean transferInProgress = new AtomicBoolean(true);
- final AtomicBoolean transferInProgress = new AtomicBoolean(true);
+ try {
- try {
+ logger.info("Stating transfer {}", transferId);
+ long start = System.currentTimeMillis();
- long start = System.currentTimeMillis();
+ onStatusCallback.accept(transferId, new TransferState()
+ .setPercentage(0)
+ .setState("RUNNING")
+ .setUpdateTimeMils(System.currentTimeMillis())
+ .setDescription("Transfer is ongoing"));
- onStatusCallback.accept(transferId, new TransferState()
- .setPercentage(0)
- .setState("RUNNING")
- .setUpdateTimeMils(System.currentTimeMillis())
- .setDescription("Transfer is ongoing"));
+ Optional<IncomingStreamingConnector> inStreamingConnectorOp = ConnectorResolver
+ .resolveIncomingStreamingConnector(request.getSourceType());
+ Optional<OutgoingStreamingConnector> outStreamingConnectorOp = ConnectorResolver
+ .resolveOutgoingStreamingConnector(request.getDestinationType());
- Optional<IncomingStreamingConnector> inStreamingConnectorOp = ConnectorResolver
- .resolveIncomingStreamingConnector(request.getSourceType());
- Optional<OutgoingStreamingConnector> outStreamingConnectorOp = ConnectorResolver
- .resolveOutgoingStreamingConnector(request.getDestinationType());
+ Optional<IncomingChunkedConnector> inChunkedConnectorOp = ConnectorResolver
+ .resolveIncomingChunkedConnector(request.getSourceType());
+ Optional<OutgoingChunkedConnector> outChunkedConnectorOp = ConnectorResolver
+ .resolveOutgoingChunkedConnector(request.getDestinationType());
- Optional<IncomingChunkedConnector> inChunkedConnectorOp = ConnectorResolver
- .resolveIncomingChunkedConnector(request.getSourceType());
- Optional<OutgoingChunkedConnector> outChunkedConnectorOp = ConnectorResolver
- .resolveOutgoingChunkedConnector(request.getDestinationType());
+ // Give priority for chunked transfers.
+ // TODO: Provide a preference at the API level
+ if (inChunkedConnectorOp.isPresent() && outChunkedConnectorOp.isPresent()) {
- // Give priority for chunked transfers.
- // TODO: Provide a preference at the API level
- if (inChunkedConnectorOp.isPresent() && outChunkedConnectorOp.isPresent()) {
+ logger.info("Starting the chunked transfer for transfer {}", transferId);
- logger.info("Starting the chunked transfer for transfer {}", transferId);
+ long chunkSize = chunkedSize * 1024 * 1024L;
- long chunkSize = chunkedSize * 1024 * 1024L;
+ CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(chunkedExecutorService);
- ExecutorService chunkedExecutorService = Executors.newFixedThreadPool(concurrentChunkedThreads);
+ long fileLength = srcCC.getMetadata().getResourceSize();
+ long uploadLength = 0L;
+ int chunkIdx = 0;
- CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(chunkedExecutorService);
+ IncomingChunkedConnector inConnector = inChunkedConnectorOp
+ .orElseThrow(() -> new Exception("Could not find an in chunked connector for type " + request.getSourceType()));
- long fileLength = srcCC.getMetadata().getResourceSize();
- long uploadLength = 0L;
- int chunkIdx = 0;
+ OutgoingChunkedConnector outConnector = outChunkedConnectorOp
+ .orElseThrow(() -> new Exception("Could not find an out chunked connector for type " + request.getDestinationType()));
- IncomingChunkedConnector inConnector = inChunkedConnectorOp
- .orElseThrow(() -> new Exception("Could not find an in chunked connector for type " + request.getSourceType()));
+ inConnector.init(srcCC);
+ outConnector.init(dstCC);
- OutgoingChunkedConnector outConnector = outChunkedConnectorOp
- .orElseThrow(() -> new Exception("Could not find an out chunked connector for type " + request.getDestinationType()));
+ while(uploadLength < fileLength) {
- inConnector.init(srcCC);
- outConnector.init(dstCC);
+ long endPos = uploadLength + chunkSize;
+ if (endPos > fileLength) {
+ endPos = fileLength;
+ }
- while(uploadLength < fileLength) {
+ String tempFile = tempDataDir + File.separator + transferId + "-" + chunkIdx;
+ completionService.submit(new ChunkMover(inConnector, outConnector, uploadLength, endPos, chunkIdx, tempFile));
- long endPos = uploadLength + chunkSize;
- if (endPos > fileLength) {
- endPos = fileLength;
- }
+ uploadLength = endPos;
+ chunkIdx++;
+ }
- String tempFile = tempDataDir + File.separator + transferId + "-" + chunkIdx;
- completionService.submit(new ChunkMover(inConnector, outConnector, uploadLength, endPos, chunkIdx, tempFile));
- uploadLength = endPos;
- chunkIdx++;
- }
+ for (int i = 0; i < chunkIdx; i++) {
+ Future<Integer> future = completionService.take();
+ }
+ inConnector.complete();
+ outConnector.complete();
+ logger.info("Completed chunked transfer for transfer {}", transferId);
- for (int i = 0; i < chunkIdx; i++) {
- Future<Integer> future = completionService.take();
- }
+ } else if (inStreamingConnectorOp.isPresent() && outStreamingConnectorOp.isPresent()) {
- inConnector.complete();
- outConnector.complete();
- logger.info("Completed chunked transfer for transfer {}", transferId);
+ logger.info("Starting streaming transfer for transfer {}", transferId);
+ IncomingStreamingConnector inConnector = inStreamingConnectorOp
+ .orElseThrow(() -> new Exception("Could not find an in streaming connector for type " + request.getSourceType()));
- } else if (inStreamingConnectorOp.isPresent() && outStreamingConnectorOp.isPresent()) {
+ OutgoingStreamingConnector outConnector = outStreamingConnectorOp
+ .orElseThrow(() -> new Exception("Could not find an out streaming connector for type " + request.getDestinationType()));
- logger.info("Starting streaming transfer for transfer {}", transferId);
- IncomingStreamingConnector inConnector = inStreamingConnectorOp
- .orElseThrow(() -> new Exception("Could not find an in streaming connector for type " + request.getSourceType()));
+ inConnector.init(srcCC);
+ outConnector.init(dstCC);
- OutgoingStreamingConnector outConnector = outStreamingConnectorOp
- .orElseThrow(() -> new Exception("Could not find an out streaming connector for type " + request.getDestinationType()));
+ String srcChild = request.getSourceChildResourcePath();
+ String dstChild = request.getDestinationChildResourcePath();
- inConnector.init(srcCC);
- outConnector.init(dstCC);
+ InputStream inputStream = srcChild.equals("") ? inConnector.fetchInputStream() : inConnector.fetchInputStream(srcChild);
+ OutputStream outputStream = dstChild.equals("") ? outConnector.fetchOutputStream() : outConnector.fetchOutputStream(dstChild);
- String srcChild = request.getSourceChildResourcePath();
- String dstChild = request.getDestinationChildResourcePath();
+ long count = 0;
+ final AtomicLong countAtomic = new AtomicLong();
+ countAtomic.set(count);
- InputStream inputStream = srcChild.equals("") ? inConnector.fetchInputStream() : inConnector.fetchInputStream(srcChild);
- OutputStream outputStream = dstChild.equals("") ? outConnector.fetchOutputStream() : outConnector.fetchOutputStream(dstChild);
+ monitorPool.submit(() -> {
+ while (true) {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ // Ignore
+ }
+ if (!transferInProgress.get()) {
+ logger.info("Status monitor is exiting for transfer {}", transferId);
+ break;
+ }
+ double transferPercentage = countAtomic.get() * 100.0 / srcCC.getMetadata().getResourceSize();
+ logger.info("Transfer percentage for transfer {} {}", transferId, transferPercentage);
+ onStatusCallback.accept(transferId, new TransferState()
+ .setPercentage(transferPercentage)
+ .setState("RUNNING")
+ .setUpdateTimeMils(System.currentTimeMillis())
+ .setDescription("Transfer Progress Updated"));
+ }
+ });
- long count = 0;
- final AtomicLong countAtomic = new AtomicLong();
+ int n;
+ byte[] buffer = new byte[128 * 1024];
+ for (count = 0L; -1 != (n = inputStream.read(buffer)); count += (long) n) {
+ outputStream.write(buffer, 0, n);
countAtomic.set(count);
+ }
- monitorPool.submit(() -> {
- while (true) {
- try {
- Thread.sleep(2000);
- } catch (InterruptedException e) {
- // Ignore
- }
- if (!transferInProgress.get()) {
- logger.info("Status monitor is exiting for transfer {}", transferId);
- break;
- }
- double transferPercentage = countAtomic.get() * 100.0 / srcCC.getMetadata().getResourceSize();
- logger.info("Transfer percentage for transfer {} {}", transferId, transferPercentage);
- onStatusCallback.accept(transferId, new TransferState()
- .setPercentage(transferPercentage)
- .setState("RUNNING")
- .setUpdateTimeMils(System.currentTimeMillis())
- .setDescription("Transfer Progress Updated"));
- }
- });
-
- int n;
- byte[] buffer = new byte[128 * 1024];
- for (count = 0L; -1 != (n = inputStream.read(buffer)); count += (long) n) {
- outputStream.write(buffer, 0, n);
- countAtomic.set(count);
- }
+ inConnector.complete();
+ outConnector.complete();
- inConnector.complete();
- outConnector.complete();
+ logger.info("Completed streaming ransfer for transfer {}", transferId);
- logger.info("Completed s ttreaming ransfer for transfer {}", transferId);
+ } else {
+ throw new Exception("No matching connector found to perform the transfer");
+ }
- } else {
- throw new Exception("No matching connector found to perform the transfer");
- }
+ long time = (System.currentTimeMillis() - start) / 1000;
- long time = (System.currentTimeMillis() - start) / 1000;
+ logger.info("Transfer {} completed. Time {} S. Speed {} MB/s", transferId, time,
+ (srcCC.getMetadata().getResourceSize() * 1.0 / time) / (1024 * 1024));
- logger.info("Transfer {} completed. Time {} S. Speed {} MB/s", transferId, time,
- (srcCC.getMetadata().getResourceSize() * 1.0 / time) / (1024 * 1024));
+ onStatusCallback.accept(transferId, new TransferState()
+ .setPercentage(100)
+ .setState("COMPLETED")
+ .setUpdateTimeMils(System.currentTimeMillis())
+ .setDescription("Transfer successfully completed"));
- onStatusCallback.accept(transferId, new TransferState()
- .setPercentage(100)
- .setState("COMPLETED")
- .setUpdateTimeMils(System.currentTimeMillis())
- .setDescription("Transfer successfully completed"));
+ exitingCallback.accept(transferId, true);
+ } catch (Exception e) {
- exitingCallback.accept(transferId, true);
- } catch (Exception e) {
+ logger.error("Transfer {} failed with error", transferId, e);
- logger.error("Transfer {} failed with error", transferId, e);
+ onStatusCallback.accept(transferId, new TransferState()
+ .setPercentage(0)
+ .setState("FAILED")
+ .setUpdateTimeMils(System.currentTimeMillis())
+ .setDescription("Transfer failed due to " + ExceptionUtils.getStackTrace(e)));
+ } finally {
+ transferInProgress.set(false);
+ }
- onStatusCallback.accept(transferId, new TransferState()
- .setPercentage(0)
- .setState("FAILED")
- .setUpdateTimeMils(System.currentTimeMillis())
- .setDescription("Transfer failed due to " + ExceptionUtils.getStackTrace(e)));
- } finally {
- transferInProgress.set(false);
- }
- });
}
public void destroy() {
- executor.shutdown();
monitorPool.shutdown();
}
@@ -230,16 +225,16 @@ public class TransportMediator {
IncomingChunkedConnector downloader;
OutgoingChunkedConnector uploader;
- long uploadLength;
+ long startPos;
long endPos;
int chunkIdx;
String tempFile;
- public ChunkMover(IncomingChunkedConnector downloader, OutgoingChunkedConnector uploader, long uploadLength,
+ public ChunkMover(IncomingChunkedConnector downloader, OutgoingChunkedConnector uploader, long startPos,
long endPos, int chunkIdx, String tempFile) {
this.downloader = downloader;
this.uploader = uploader;
- this.uploadLength = uploadLength;
+ this.startPos = startPos;
this.endPos = endPos;
this.chunkIdx = chunkIdx;
this.tempFile = tempFile;
@@ -247,8 +242,8 @@ public class TransportMediator {
@Override
public Integer call() throws Exception {
- downloader.downloadChunk(chunkIdx, uploadLength, endPos, tempFile);
- uploader.uploadChunk(chunkIdx, uploadLength, endPos, tempFile);
+ downloader.downloadChunk(chunkIdx, startPos, endPos, tempFile);
+ uploader.uploadChunk(chunkIdx, startPos, endPos, tempFile);
new File(tempFile).delete();
return chunkIdx;
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/IncomingChunkedConnector.java b/core/src/main/java/org/apache/airavata/mft/core/api/IncomingChunkedConnector.java
index 37288c6..24908a2 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/IncomingChunkedConnector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/IncomingChunkedConnector.java
@@ -1,5 +1,7 @@
package org.apache.airavata.mft.core.api;
+import java.io.InputStream;
+
public interface IncomingChunkedConnector extends BasicConnector {
public void downloadChunk(int chunkId, long startByte, long endByte, String downloadFile) throws Exception;
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/OutgoingChunkedConnector.java b/core/src/main/java/org/apache/airavata/mft/core/api/OutgoingChunkedConnector.java
index 9a2f3cb..d9cc5b5 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/OutgoingChunkedConnector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/OutgoingChunkedConnector.java
@@ -1,5 +1,7 @@
package org.apache.airavata.mft.core.api;
+import java.io.InputStream;
+
public interface OutgoingChunkedConnector extends BasicConnector {
public void uploadChunk(int chunkId, long startByte, long endByte, String uploadFile) throws Exception;
}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3IncomingConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3IncomingConnector.java
index be2c9ee..29e8263 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3IncomingConnector.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3IncomingConnector.java
@@ -8,6 +8,7 @@ import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.S3Object;
+import com.amazonaws.util.IOUtils;
import org.apache.airavata.mft.core.api.ConnectorConfig;
import org.apache.airavata.mft.core.api.IncomingChunkedConnector;
import org.apache.airavata.mft.core.api.IncomingStreamingConnector;
@@ -23,6 +24,7 @@ import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.InputStream;
@@ -88,7 +90,7 @@ public class S3IncomingConnector implements IncomingChunkedConnector, IncomingSt
resource.getFile().getResourcePath());
rangeObjectRequest.setRange(startByte, endByte - 1);
ObjectMetadata objectMetadata = s3Client.getObject(rangeObjectRequest, new File(downloadFile));
- logger.info("Downloaded S3 chunk to path {} for resource id {}", downloadFile, resource.getResourceId());
+ logger.debug("Downloaded S3 chunk to path {} for resource id {}", downloadFile, resource.getResourceId());
}
@Override
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3OutgoingConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3OutgoingConnector.java
index c7be388..aa59378 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3OutgoingConnector.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3OutgoingConnector.java
@@ -21,6 +21,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
+import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
@@ -92,7 +93,7 @@ public class S3OutgoingConnector implements OutgoingChunkedConnector {
UploadPartResult uploadResult = s3Client.uploadPart(uploadRequest);
this.partETags.add(uploadResult.getPartETag());
- logger.info("Uploaded S3 chunk to path {} for resource id {}", uploadFile, resource.getResourceId());
+ logger.debug("Uploaded S3 chunk to path {} for resource id {}", uploadFile, resource.getResourceId());
}