forked from nextflow-io/nextflow
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Improve S3 file upload/download via Transfer manager
This commit allows the use of the AWS Transfer Manager when when copying S3 files from and to an S3 bucket. The use of the "native" transfer provides better performance and stability Signed-off-by: Jorge Aguilera <[email protected]> Signed-off-by: Paolo Di Tommaso <[email protected]> Co-authored-by: Paolo Di Tommaso <[email protected]>
- Loading branch information
1 parent
1c23b40
commit 7e8d2a5
Showing
24 changed files
with
993 additions
and
247 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
110 changes: 110 additions & 0 deletions
110
modules/nextflow/src/main/groovy/nextflow/file/FileTransferPool.groovy
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| /* | ||
| * Copyright 2020-2022, Seqera Labs | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| * | ||
| */ | ||
|
|
||
| package nextflow.file | ||
|
|
||
| import java.util.concurrent.ExecutorService | ||
| import java.util.concurrent.ThreadPoolExecutor | ||
|
|
||
| import groovy.transform.CompileStatic | ||
| import groovy.transform.Memoized | ||
| import groovy.util.logging.Slf4j | ||
| import nextflow.Global | ||
| import nextflow.util.Duration | ||
| import nextflow.util.ThreadPoolBuilder | ||
| import nextflow.util.ThreadPoolHelper | ||
|
|
||
| /** | ||
| * Holder object for file transfrer thread pool | ||
| * | ||
| * @author Paolo Di Tommaso <[email protected]> | ||
| */ | ||
| @Slf4j | ||
| @CompileStatic | ||
| class FileTransferPool { | ||
|
|
||
| final static private DEFAULT_MIN_THREAD = 1 | ||
| final static private DEFAULT_MAX_THREAD = Math.min(Runtime.runtime.availableProcessors()*2, 10) | ||
| final static private DEFAULT_QUEUE = 10_000 | ||
| final static private DEFAULT_KEEP_ALIVE = Duration.of('60sec') | ||
| final DEFAULT_MAX_AWAIT = Duration.of('12 hour') | ||
|
|
||
| final private Integer minThreads | ||
| final private Integer maxThreads | ||
| final private Integer maxQueueSize | ||
| final private Duration keepAlive | ||
| final private Boolean allowThreadTimeout | ||
| final private Duration maxAwait | ||
| final private ThreadPoolExecutor executorService | ||
|
|
||
| static private FileTransferPool instance0 | ||
|
|
||
| FileTransferPool(Map config) { | ||
| this.minThreads = config.navigate("threadPool.FileTransfer.minThreads", DEFAULT_MIN_THREAD) as Integer | ||
| this.maxThreads = config.navigate("threadPool.FileTransfer.maxThreads", DEFAULT_MAX_THREAD) as Integer | ||
| this.maxQueueSize = config.navigate("threadPool.FileTransfer.maxQueueSize", DEFAULT_QUEUE) as Integer | ||
| this.keepAlive = config.navigate("threadPool.FileTransfer.keepAlive", DEFAULT_KEEP_ALIVE) as Duration | ||
| this.allowThreadTimeout = config.navigate("threadPool.FileTransfer.allowThreadTimeout", false) as Boolean | ||
| this.maxAwait = config.navigate("threadPool.FileTransfer.maxAwait", DEFAULT_MAX_AWAIT) as Duration | ||
|
|
||
| if( minThreads>maxThreads ) { | ||
| log.debug("FileTransfer minThreads ($minThreads) cannot be greater than maxThreads ($maxThreads) - Setting minThreads to $maxThreads") | ||
| minThreads = maxThreads | ||
| } | ||
|
|
||
| executorService = new ThreadPoolBuilder() | ||
| .withName('FileTransfer') | ||
| .withMinSize(minThreads) | ||
| .withMaxSize(maxThreads) | ||
| .withQueueSize(maxQueueSize) | ||
| .withKeepAliveTime(keepAlive) | ||
| .withAllowCoreThreadTimeout(allowThreadTimeout) | ||
| .build() | ||
| } | ||
|
|
||
| private ExecutorService getExecutorService0() { | ||
| return executorService | ||
| } | ||
|
|
||
| private void shutdown0(boolean hard) { | ||
| if( hard ) { | ||
| executorService.shutdownNow() | ||
| return | ||
| } | ||
|
|
||
| executorService.shutdown() | ||
| // wait for ongoing file transfer to complete | ||
| final waitMsg = "Waiting files transfer to complete (%d files)" | ||
| final exitMsg = "Exiting before FileTransfer thread pool complete -- Some files maybe lost" | ||
| ThreadPoolHelper.await(executorService, maxAwait, waitMsg, exitMsg) | ||
| } | ||
|
|
||
| @Memoized | ||
| static synchronized ExecutorService getExecutorService() { | ||
| final session = Global.session | ||
| if( session == null ) | ||
| throw new IllegalStateException("Nextflow session object has not been created yet") | ||
| instance0 = new FileTransferPool(session.getConfig()) | ||
| return instance0.getExecutorService0() | ||
| } | ||
|
|
||
| static shutdown(boolean hard) { | ||
| if( instance0 ) | ||
| instance0.shutdown0(hard) | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
66 changes: 66 additions & 0 deletions
66
modules/nextflow/src/main/groovy/nextflow/util/ThreadPoolHelper.groovy
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,66 @@ | ||
| /* | ||
| * Copyright 2020-2022, Seqera Labs | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| * | ||
| */ | ||
|
|
||
| package nextflow.util | ||
|
|
||
| import java.util.concurrent.ExecutorService | ||
| import java.util.concurrent.ThreadPoolExecutor | ||
| import java.util.concurrent.TimeUnit | ||
|
|
||
| import groovy.transform.CompileStatic | ||
| import groovy.util.logging.Slf4j | ||
| /** | ||
| * Thread pool helpers | ||
| * | ||
| * @author Paolo Di Tommaso <[email protected]> | ||
| */ | ||
| @CompileStatic | ||
| @Slf4j | ||
| class ThreadPoolHelper { | ||
|
|
||
| static void await(ExecutorService pool, Duration maxAwait, String waitMessage, String exitMsg) { | ||
| final max = maxAwait.millis | ||
| final t0 = System.currentTimeMillis() | ||
| // wait for ongoing file transfer to complete | ||
| int count=0 | ||
| while( true ) { | ||
| final terminated = pool.awaitTermination(5, TimeUnit.SECONDS) | ||
| if( terminated ) | ||
| break | ||
|
|
||
| final delta = System.currentTimeMillis()-t0 | ||
| if( delta > max ) { | ||
| log.warn(exitMsg) | ||
| break | ||
| } | ||
|
|
||
| final p1 = ((ThreadPoolExecutor)pool) | ||
| final pending = p1.getTaskCount() - p1.getCompletedTaskCount() | ||
| // log to console every 10 minutes (120 * 5 sec) | ||
| if( count % 120 == 0 ) { | ||
| log.info1(String.format(waitMessage, pending)) | ||
| } | ||
| // log to the debug file every minute (12 * 5 sec) | ||
| else if( count % 12 == 0 ) { | ||
| log.debug(String.format(waitMessage, pending)) | ||
| } | ||
| // increment the count | ||
| count++ | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.