binding = super.makeBinding()
+
+ binding.K8sResolveSymlinks = null
+
+ if ( binding.stage_inputs && storage && localWorkDir ) {
+ final String cmd = """\
+ # create symlinks
+ if test -f "${workDir.toString()}/.command.symlinks"; then
+ bash "${workDir.toString()}/.command.symlinks" || true
+ fi
+ """.stripIndent()
+ binding.stage_inputs = cmd + binding.stage_inputs
+ }
+
+ if ( localWorkDir ) {
+ binding.unstage_outputs = copyStrategy.getUnstageOutputFilesScript(outputFiles, localWorkDir)
+ }
+
+ return binding
+ }
+
+ @Override
+ protected String getLaunchCommand(String interpreter, String env) {
+ assert storage && localWorkDir
+ String cmd = ''
+ cmd += "local INFILESTIME=\$(\"/etc/nextflow/${statFileName}\" infiles \"${workDir.toString()}/.command.infiles\" \"${getStorageLocalWorkDir()}\" \"\$PWD/\" || true)\n"
+ cmd += super.getLaunchCommand(interpreter, env)
+ if( isTraceRequired() ){
+ cmd += "\nlocal exitCode=\$?"
+ cmd += """\necho \"infiles_time=\${INFILESTIME}" >> ${workDir.resolve(TaskRun.CMD_TRACE)}\n"""
+ cmd += "return \$exitCode\n"
+ }
+ return cmd
+ }
+
+ @Override
+ String getCleanupCmd(String scratch) {
+ assert storage && localWorkDir
+ String cmd = super.getCleanupCmd( scratch )
+ cmd += "mkdir -p \"${localWorkDir.toString()}/\" || true\n"
+ cmd += "local OUTFILESTIME=\$(\"/etc/nextflow/${statFileName}\" outfiles \"${workDir.toString()}/.command.outfiles\" \"${getStorageLocalWorkDir()}\" \"${localWorkDir.toString()}/\" || true)\n"
+ if ( isTraceRequired() ) {
+ cmd += "echo \"outfiles_time=\${OUTFILESTIME}\" >> ${workDir.resolve(TaskRun.CMD_TRACE)}"
+ }
+ return cmd
+ }
+
+}
\ No newline at end of file
diff --git a/plugins/nf-cws/src/main/nextflow/cws/processor/CWSTaskPollingMonitor.groovy b/plugins/nf-cws/src/main/nextflow/cws/processor/CWSTaskPollingMonitor.groovy
index 7784618..a908936 100644
--- a/plugins/nf-cws/src/main/nextflow/cws/processor/CWSTaskPollingMonitor.groovy
+++ b/plugins/nf-cws/src/main/nextflow/cws/processor/CWSTaskPollingMonitor.groovy
@@ -1,11 +1,20 @@
package nextflow.cws.processor
+import groovy.transform.CompileStatic
import groovy.util.logging.Slf4j
import nextflow.Session
+import nextflow.cws.CWSConfig
+import nextflow.cws.wow.file.OfflineLocalPath
+import nextflow.cws.wow.file.WOWFileAttributes
+import nextflow.cws.wow.file.WorkdirPath
+import nextflow.cws.wow.util.LocalFileWalker
+import nextflow.processor.PublishDir
+import nextflow.processor.TaskHandler
import nextflow.processor.TaskPollingMonitor
import nextflow.util.Duration
@Slf4j
+@CompileStatic
class CWSTaskPollingMonitor extends TaskPollingMonitor {
/**
@@ -13,6 +22,8 @@ class CWSTaskPollingMonitor extends TaskPollingMonitor {
*/
private final SchedulerBatch schedulerBatch
+ private final CWSConfig cwsConfig
+
/**
* Create the task polling monitor with the provided named parameters object.
*
@@ -28,6 +39,7 @@ class CWSTaskPollingMonitor extends TaskPollingMonitor {
protected CWSTaskPollingMonitor(Map params) {
super(params)
this.schedulerBatch = params.schedulerBatch as SchedulerBatch
+ cwsConfig = new CWSConfig(session.config.navigate('cws') as Map)
}
static TaskPollingMonitor create(Session session, String name, int defQueueSize, Duration defPollInterval, SchedulerBatch schedulerBatch ) {
@@ -54,4 +66,38 @@ class CWSTaskPollingMonitor extends TaskPollingMonitor {
return pendingTasks
}
+ private static void checkPublishDirMode(TaskHandler handler ) {
+ def publishDirs = handler.task.config.get('publishDir')
+ if ( publishDirs && publishDirs instanceof List ) {
+ for( Object params : publishDirs ) {
+ if( !params ) continue
+ if( params instanceof Map ) {
+ def mode = PublishDir.create(params).getMode()
+ // We only support COPY and MOVE, if the user uses a different mode, we set it to COPY
+ if ( !(mode in [ PublishDir.Mode.COPY, PublishDir.Mode.MOVE]) ) {
+ params.mode = PublishDir.Mode.COPY
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ protected void finalizeTask(TaskHandler handler) {
+ if (!cwsConfig.strategyIsLocationAware()) {
+ super.finalizeTask(handler)
+ return
+ }
+ def workDir = handler.task.workDir
+ def helper = LocalFileWalker.createWorkdirHelper( workDir )
+ if ( helper ) {
+ def attributes = new WOWFileAttributes(workDir)
+ OfflineLocalPath path = new WorkdirPath( workDir, attributes, workDir, helper )
+ handler.task.workDir = path
+ }
+ checkPublishDirMode(handler)
+ super.finalizeTask(handler)
+ helper?.validate()
+ }
+
}
diff --git a/plugins/nf-cws/src/main/nextflow/cws/processor/SchedulerBatch.groovy b/plugins/nf-cws/src/main/nextflow/cws/processor/SchedulerBatch.groovy
index ae46ddb..480e906 100644
--- a/plugins/nf-cws/src/main/nextflow/cws/processor/SchedulerBatch.groovy
+++ b/plugins/nf-cws/src/main/nextflow/cws/processor/SchedulerBatch.groovy
@@ -1,8 +1,12 @@
package nextflow.cws.processor
+import groovy.transform.CompileStatic
+
+@CompileStatic
abstract class SchedulerBatch {
private final int batchSize
+
private int currentlySubmitted = 0
SchedulerBatch( int batchSize ) {
diff --git a/plugins/nf-cws/src/main/nextflow/cws/wow/file/LocalFile.groovy b/plugins/nf-cws/src/main/nextflow/cws/wow/file/LocalFile.groovy
new file mode 100644
index 0000000..730bd6a
--- /dev/null
+++ b/plugins/nf-cws/src/main/nextflow/cws/wow/file/LocalFile.groovy
@@ -0,0 +1,22 @@
+package nextflow.cws.wow.file
+
+import groovy.transform.CompileStatic
+
+import java.nio.file.Path
+
+@CompileStatic
+class LocalFile extends File {
+
+ private final LocalPath localPath
+
+ LocalFile( LocalPath localPath ){
+ super( localPath.toString() )
+ this.localPath = localPath
+ }
+
+ @Override
+ Path toPath() {
+ return localPath
+ }
+
+}
\ No newline at end of file
diff --git a/plugins/nf-cws/src/main/nextflow/cws/wow/file/LocalPath.groovy b/plugins/nf-cws/src/main/nextflow/cws/wow/file/LocalPath.groovy
new file mode 100644
index 0000000..57cbe87
--- /dev/null
+++ b/plugins/nf-cws/src/main/nextflow/cws/wow/file/LocalPath.groovy
@@ -0,0 +1,227 @@
+package nextflow.cws.wow.file
+
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+import nextflow.cws.wow.filesystem.WOWFileSystem
+
+import java.nio.file.*
+
+@Slf4j
+@CompileStatic
+class LocalPath implements Path, Serializable {
+
+ protected final Path path
+
+ private transient final WOWFileAttributes attributes
+
+ boolean createdSymlinks = false
+
+ protected Path workDir
+
+ protected LocalPath(Path path, WOWFileAttributes attributes, Path workDir ) {
+ this.path = path
+ this.attributes = attributes
+ this.workDir = workDir
+ }
+
+ Path getInner() {
+ path
+ }
+
+ LocalPath toLocalPath( Path path, WOWFileAttributes attributes = null ){
+ toLocalPath( path, attributes, workDir )
+ }
+
+ static LocalPath toLocalPath( Path path, WOWFileAttributes attributes, Path workDir ){
+ ( path instanceof LocalPath ) ? path as LocalPath : new LocalPath( path, attributes, workDir )
+ }
+
+ T asType( Class c ) {
+ if ( c.isAssignableFrom( getClass() ) ) return (T) this
+ if ( c.isAssignableFrom( LocalPath.class ) ) return (T) toFile()
+ if ( c == String.class ) return (T) toString()
+ log.debug("Invoke method asType $c on ${this.class}")
+ return super.asType( c )
+ }
+
+ String getBaseName() {
+ path.getBaseName()
+ }
+
+ boolean isDirectory( LinkOption... options ) {
+ attributes ? attributes.isDirectory() : 0
+ }
+
+ long size() {
+ attributes ? attributes.size() : 0
+ }
+
+ boolean empty(){
+ this.size() == 0
+ }
+
+ boolean asBoolean(){
+ true
+ }
+
+ @Override
+ FileSystem getFileSystem() {
+ WOWFileSystem.INSTANCE
+ }
+
+ @Override
+ boolean isAbsolute() {
+ path.isAbsolute()
+ }
+
+ @Override
+ Path getRoot() {
+ path.getRoot()
+ }
+
+ @Override
+ Path getFileName() {
+ path.getFileName()
+ }
+
+ @Override
+ Path getParent() {
+ toLocalPath( path.getParent() )
+ }
+
+ @Override
+ int getNameCount() {
+ path.getNameCount()
+ }
+
+ @Override
+ Path getName(int index) {
+ path.getName( index )
+ }
+
+ @Override
+ Path subpath(int beginIndex, int endIndex) {
+ toLocalPath( path.subpath( beginIndex, endIndex ) )
+ }
+
+ @Override
+ boolean startsWith(Path other) {
+ path.startsWith( other )
+ }
+
+ @Override
+ boolean startsWith(String other) {
+ path.startsWith(other)
+ }
+
+ @Override
+ boolean endsWith(Path other) {
+ path.endsWith( other )
+ }
+
+ @Override
+ boolean endsWith(String other) {
+ path.endsWith( other )
+ }
+
+ @Override
+ Path normalize() {
+ toLocalPath( path.normalize(), attributes )
+ }
+
+ @Override
+ Path resolve(Path other) {
+ toLocalPath( path.resolve( other ), new WOWFileAttributes( other ) )
+ }
+
+ @Override
+ Path resolve(String other) {
+ resolve( Path.of( other ) )
+ }
+
+ @Override
+ Path resolveSibling(Path other) {
+ path.resolveSibling( other )
+ }
+
+ @Override
+ Path resolveSibling(String other) {
+ path.resolveSibling( other )
+ }
+
+ @Override
+ Path relativize(Path other) {
+ if ( other instanceof LocalPath ){
+ def localPath = (LocalPath) other
+ return toLocalPath( path.relativize( localPath.path), (WOWFileAttributes) localPath.attributes, localPath.workDir )
+ }
+ path.relativize( other )
+ }
+
+ @Override
+ URI toUri() {
+ getFileSystem().provider().getScheme() + "://" + path.toAbsolutePath() as URI
+ }
+
+ String toUriString() {
+ getFileSystem().provider().getScheme() + ":/" + path.toAbsolutePath()
+ }
+
+ Path toAbsolutePath(){
+ toLocalPath( path.toAbsolutePath(), attributes )
+ }
+
+ @Override
+ Path toRealPath(LinkOption... options) throws IOException {
+ attributes.destination ? toLocalPath( attributes.destination ) : toLocalPath( path.toRealPath( options ) )
+ }
+
+ @Override
+ File toFile() {
+ new LocalFile( this )
+ }
+
+ @Override
+ WatchKey register(WatchService watcher, WatchEvent.Kind[] events, WatchEvent.Modifier... modifiers) throws IOException {
+ path.register( watcher, events, modifiers )
+ }
+
+ @Override
+ WatchKey register(WatchService watcher, WatchEvent.Kind... events) throws IOException {
+ path.register( watcher, events )
+ }
+
+ @Override
+ int compareTo(Path other) {
+ if ( other instanceof LocalPath ){
+ return path <=> ((LocalPath) other).path
+ }
+ path <=> other
+ }
+
+ @Override
+ String toString() {
+ path.toString()
+ }
+
+ WOWFileAttributes getAttributes(){
+ attributes
+ }
+
+ @Override
+ boolean equals(Object obj) {
+ if ( obj instanceof LocalPath ){
+ return path == ((LocalPath) obj).path
+ }
+ if ( obj instanceof Path ){
+ return path == obj
+ }
+ return false
+ }
+
+ @Override
+ int hashCode() {
+ path.hashCode() * 2 + 1
+ }
+
+}
\ No newline at end of file
diff --git a/plugins/nf-cws/src/main/nextflow/cws/wow/file/OfflineLocalPath.groovy b/plugins/nf-cws/src/main/nextflow/cws/wow/file/OfflineLocalPath.groovy
new file mode 100644
index 0000000..ae50514
--- /dev/null
+++ b/plugins/nf-cws/src/main/nextflow/cws/wow/file/OfflineLocalPath.groovy
@@ -0,0 +1,20 @@
+package nextflow.cws.wow.file
+
+import groovy.transform.CompileStatic
+
+import java.nio.file.Path
+
+/**
+ * We need this class to be able to iterate through a subdirectory of a workdir
+ */
+@CompileStatic
+class OfflineLocalPath extends LocalPath {
+
+ final WorkdirHelper workdirHelper
+
+ OfflineLocalPath(Path path, WOWFileAttributes attributes, Path workDir, WorkdirHelper workdirHelper ) {
+ super(path, attributes, workDir)
+ this.workdirHelper = workdirHelper
+ }
+
+}
diff --git a/plugins/nf-cws/src/main/nextflow/cws/wow/file/WOWFileAttributes.groovy b/plugins/nf-cws/src/main/nextflow/cws/wow/file/WOWFileAttributes.groovy
new file mode 100644
index 0000000..939fc30
--- /dev/null
+++ b/plugins/nf-cws/src/main/nextflow/cws/wow/file/WOWFileAttributes.groovy
@@ -0,0 +1,174 @@
+package nextflow.cws.wow.file
+
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+
+import java.nio.file.Path
+import java.nio.file.attribute.BasicFileAttributes
+import java.nio.file.attribute.FileTime
+
+@Slf4j
+@CompileStatic
+class WOWFileAttributes implements BasicFileAttributes {
+
+ /**
+ * Helper to parse the file attributes returned by the getStatsAndResolveSymlinks.c
+ */
+ static private final int FILE_EXISTS = 1
+ static private final int REAL_PATH = 2
+ static private final int SIZE = 3
+ static private final int FILE_TYPE = 4
+ static private final int CREATION_DATE = 5
+ static private final int ACCESS_DATE = 6
+ static private final int MODIFICATION_DATE = 7
+
+ private final boolean directory
+
+ private final boolean link
+
+ private final long size
+
+ private final String fileType
+
+ private final FileTime creationDate
+
+ private final FileTime accessDate
+
+ private final FileTime modificationDate
+
+ private final Path destination
+
+ private final boolean local
+
+ private static FileTime fileTimeFromString( String date ) {
+ for (int i = 0; i < date.length(); i++) {
+ if ( !date.charAt(i).isDigit() ) {
+ return null
+ }
+ }
+ long nanos = Long.parseLong(date)
+ return FileTime.fromMillis(nanos / 1e6 as long)
+ }
+
+ WOWFileAttributes( String[] data ) {
+ boolean fileExists = data[ FILE_EXISTS ] == "1"
+ if ( fileExists && data.length != 8 ) throw new RuntimeException( "Cannot parse row (8 columns required): ${data.join(',')}" )
+ destination = data.length > REAL_PATH && data[ REAL_PATH ] ? data[ REAL_PATH ] as Path : null
+ if ( data.length != 8 ) {
+ this.directory = false
+ this.link = true
+ this.size = 0
+ this.fileType = null
+ this.creationDate = null
+ this.accessDate = null
+ this.modificationDate = null
+ this.local = false
+ return
+ }
+ this.link = !data[ REAL_PATH ].isEmpty()
+ this.size = data[ SIZE ] as Long
+ String fileType = data[ FILE_TYPE ]
+ this.accessDate = fileTimeFromString(data[ ACCESS_DATE ])
+ this.modificationDate = fileTimeFromString(data[ MODIFICATION_DATE ])
+ this.creationDate = fileTimeFromString(data[ CREATION_DATE ]) ?: this.modificationDate
+ if ( fileType.startsWith("non-local ") ) {
+ this.local = false
+ this.fileType = fileType.substring( 10 )
+ } else {
+ this.local = true
+ this.fileType = fileType
+ }
+ this.directory = this.fileType == 'directory'
+ if ( !directory && !this.fileType.contains( 'file' ) ){
+ log.error( "Unknown type: ${this.fileType}" )
+ }
+ }
+
+ WOWFileAttributes(Path path ) {
+ if ( !path.exists() ) {
+ directory = true
+ link = false
+ size = 4096
+ fileType = 'directory'
+ creationDate = FileTime.fromMillis( 0 )
+ accessDate = FileTime.fromMillis( 0 )
+ modificationDate = FileTime.fromMillis( 0 )
+ destination = path
+ local = false
+ } else if ( path.isDirectory() ) {
+ directory = true
+ link = false
+ size = 4096
+ fileType = 'directory'
+ creationDate = FileTime.fromMillis(0)
+ accessDate = FileTime.fromMillis(0)
+ modificationDate = FileTime.fromMillis(0)
+ destination = path
+ local = false
+ } else {
+ directory = false
+ link = path.isLink()
+ size = path.size()
+ fileType = link ? 'symbolic link' : 'regular file'
+ creationDate = null
+ accessDate = null
+ modificationDate = FileTime.fromMillis(path.lastModified())
+ destination = link ? path.toRealPath() : path
+ local = !link
+ }
+ }
+
+ @Override
+ FileTime lastModifiedTime() {
+ modificationDate
+ }
+
+ @Override
+ FileTime lastAccessTime() {
+ accessDate
+ }
+
+ @Override
+ FileTime creationTime() {
+ creationDate
+ }
+
+ @Override
+ boolean isRegularFile() {
+ !directory
+ }
+
+ @Override
+ boolean isDirectory() {
+ directory
+ }
+
+ @Override
+ boolean isSymbolicLink() {
+ link
+ }
+
+ @Override
+ boolean isOther() {
+ false
+ }
+
+ @Override
+ long size() {
+ size
+ }
+
+ @Override
+ Object fileKey() {
+ null
+ }
+
+ Path getDestination(){
+ destination
+ }
+
+ boolean isLocal() {
+ local
+ }
+
+}
\ No newline at end of file
diff --git a/plugins/nf-cws/src/main/nextflow/cws/wow/file/WorkdirHelper.groovy b/plugins/nf-cws/src/main/nextflow/cws/wow/file/WorkdirHelper.groovy
new file mode 100644
index 0000000..3b3420d
--- /dev/null
+++ b/plugins/nf-cws/src/main/nextflow/cws/wow/file/WorkdirHelper.groovy
@@ -0,0 +1,80 @@
+package nextflow.cws.wow.file
+
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+
+import java.nio.file.DirectoryStream
+import java.nio.file.Path
+
+@Slf4j
+@CompileStatic
+class WorkdirHelper {
+
+ private final Map paths
+
+ private final Path rootPath
+
+ private boolean validated = false
+
+ WorkdirHelper( Path rootPath, Map paths) {
+ this.rootPath = rootPath
+ this.paths = paths
+ }
+
+ void validate() {
+ validated = true
+ }
+
+ boolean isValidated() {
+ return validated
+ }
+
+ LocalPath get( Path path ) {
+ if ( validated ) {
+ throw new IllegalStateException("WorkdirHelper validated")
+ }
+ paths.get( path )
+ }
+
+ Path relativeToWorkdir( LocalPath path ) {
+ new LocalPath( rootPath.relativize( path.path ), path.getAttributes(), path.workDir )
+ }
+
+ int getNameCount() {
+ rootPath.getNameCount()
+ }
+
+ DirectoryStream getDirectoryStream(Path path) {
+ if( validated ) {
+ throw new IllegalStateException("WorkdirHelper validated")
+ }
+ // If this is a local path, we need to check if the path is relative to the rootPath
+ boolean useLocalPath = path.startsWith(rootPath)
+ return new DirectoryStream() {
+ @Override
+ Iterator iterator() {
+ final def cp = path as LocalPath
+ def all = paths.entrySet().findAll {
+ Path toCompareAgainst = useLocalPath ? it.value.getInner() : it.key
+ def result = toCompareAgainst.parent == cp.getInner()
+ return result
+ }.collect {
+ (Path) it.value
+ }
+ return all.iterator()
+ }
+ @Override
+ void close() throws IOException {}
+ }
+ }
+
+ @Override
+ String toString() {
+ return "WorkdirHelper{" +
+ "paths=" + paths +
+ ", rootPath=" + rootPath +
+ ", validated=" + validated +
+ '}'
+ }
+
+}
diff --git a/plugins/nf-cws/src/main/nextflow/cws/wow/file/WorkdirPath.groovy b/plugins/nf-cws/src/main/nextflow/cws/wow/file/WorkdirPath.groovy
new file mode 100644
index 0000000..43541a3
--- /dev/null
+++ b/plugins/nf-cws/src/main/nextflow/cws/wow/file/WorkdirPath.groovy
@@ -0,0 +1,46 @@
+package nextflow.cws.wow.file
+
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+
+import java.nio.file.Path
+
+/** This special path is used to represent the workdir path in the local file system.
+ * This is required to calculate the correct relative path to the workdir.
+ */
+@Slf4j
+@CompileStatic
+class WorkdirPath extends OfflineLocalPath {
+
+ WorkdirPath(Path path, WOWFileAttributes attributes, Path workDir, WorkdirHelper workdirHelper) {
+ super(path, attributes, workDir, workdirHelper)
+ }
+
+ @Override
+ Path relativize(Path other) {
+ if ( other instanceof LocalPath ) {
+ def otherPath = ((LocalPath) other).path
+ if ( this.path == otherPath ) {
+ return Path.of("")
+ }
+ return workdirHelper.relativeToWorkdir( (LocalPath) other )
+ }
+ return this.path.relativize( other )
+ }
+
+ /**
+ * This method considers that the current path != the local path.
+ * If a file is found in the local paths, it returns the local path.
+ * @param other
+ * @return
+ */
+ Path resolve( String other ) {
+ def file = this.path.resolve( other )
+ workdirHelper.get( file ) ?: file
+ }
+
+ @Override
+ int getNameCount() {
+ return workdirHelper.getNameCount()
+ }
+}
diff --git a/plugins/nf-cws/src/main/nextflow/cws/wow/filesystem/WOWFileSystem.groovy b/plugins/nf-cws/src/main/nextflow/cws/wow/filesystem/WOWFileSystem.groovy
new file mode 100644
index 0000000..ac0a3dd
--- /dev/null
+++ b/plugins/nf-cws/src/main/nextflow/cws/wow/filesystem/WOWFileSystem.groovy
@@ -0,0 +1,82 @@
+package nextflow.cws.wow.filesystem
+
+import groovy.transform.CompileDynamic
+import groovy.transform.CompileStatic
+
+import java.nio.file.*
+import java.nio.file.attribute.UserPrincipalLookupService
+import java.nio.file.spi.FileSystemProvider
+
+@CompileStatic
+class WOWFileSystem extends FileSystem {
+
+ static final WOWFileSystem INSTANCE = new WOWFileSystem()
+
+ @Override
+ FileSystemProvider provider() {
+ WOWFileSystemProvider.INSTANCE
+ }
+
+ @Override
+ void close() throws IOException {
+ }
+
+ @Override
+ boolean isOpen() {
+ true
+ }
+
+ @Override
+ boolean isReadOnly() {
+ false
+ }
+
+ @Override
+ String getSeparator() {
+ "/"
+ }
+
+ @Override
+ Iterable getRootDirectories() {
+ throw new UnsupportedOperationException("Root directories not supported by ${provider().getScheme().toUpperCase()} file system")
+ }
+
+ @Override
+ Iterable getFileStores() {
+ throw new UnsupportedOperationException("File stores not supported by ${provider().getScheme().toUpperCase()} file system")
+ }
+
+ @Override
+ Set supportedFileAttributeViews() {
+ new HashSet<>()
+ }
+
+ @Override
+ Path getPath(String s, String... strings) {
+ throw new UnsupportedOperationException("Path get not supported by ${provider().getScheme().toUpperCase()} file system")
+ }
+
+ @Override
+ @CompileDynamic
+ PathMatcher getPathMatcher(String s) {
+ new PathMatcher() {
+ private final def matcher = FileSystems.getDefault().getPathMatcher( s )
+ @Override
+ boolean matches(Path path) {
+ // Make this a Unix Path and use the default matcher
+ return matcher.matches( Path.of(path.toString()) )
+ }
+ }
+ }
+
+ @Override
+ UserPrincipalLookupService getUserPrincipalLookupService() {
+ throw new UnsupportedOperationException("User principal lookup service not supported by ${provider().getScheme().toUpperCase()} file system")
+ }
+
+ @Override
+ WatchService newWatchService() throws IOException {
+ throw new UnsupportedOperationException("Watch service not supported by ${provider().getScheme().toUpperCase()} file system")
+ }
+
+}
diff --git a/plugins/nf-cws/src/main/nextflow/cws/wow/filesystem/WOWFileSystemPathFactory.groovy b/plugins/nf-cws/src/main/nextflow/cws/wow/filesystem/WOWFileSystemPathFactory.groovy
new file mode 100644
index 0000000..fe5731e
--- /dev/null
+++ b/plugins/nf-cws/src/main/nextflow/cws/wow/filesystem/WOWFileSystemPathFactory.groovy
@@ -0,0 +1,39 @@
+package nextflow.cws.wow.filesystem
+
+import groovy.transform.CompileStatic
+import nextflow.cws.wow.file.LocalPath
+import nextflow.file.FileSystemPathFactory
+
+import java.nio.file.Path
+
+@CompileStatic
+class WOWFileSystemPathFactory extends FileSystemPathFactory {
+
+ @Override
+ protected Path parseUri(String uri) {
+ if ( uri.startsWith("wow://") ) {
+ def of = Path.of(uri.substring(5))
+ return LocalPath.toLocalPath(of, null, null )
+ }
+ return null
+ }
+
+ @Override
+ protected String toUriString(Path path) {
+ if ( path instanceof LocalPath ) {
+ return path.toUriString()
+ }
+ return null
+ }
+
+ @Override
+ protected String getBashLib(Path target) {
+ null
+ }
+
+ @Override
+ protected String getUploadCmd(String source, Path target) {
+ throw new UnsupportedOperationException("WOWFileSystemPathFactory does not support getUploadCmd")
+ }
+
+}
diff --git a/plugins/nf-cws/src/main/nextflow/cws/wow/filesystem/WOWFileSystemProvider.groovy b/plugins/nf-cws/src/main/nextflow/cws/wow/filesystem/WOWFileSystemProvider.groovy
new file mode 100644
index 0000000..057da10
--- /dev/null
+++ b/plugins/nf-cws/src/main/nextflow/cws/wow/filesystem/WOWFileSystemProvider.groovy
@@ -0,0 +1,237 @@
+package nextflow.cws.wow.filesystem
+
+import groovy.transform.CompileStatic
+import groovy.transform.PackageScope
+import groovy.util.logging.Slf4j
+import nextflow.cws.SchedulerClient
+import nextflow.cws.wow.file.LocalPath
+import nextflow.cws.wow.file.OfflineLocalPath
+import nextflow.file.FileSystemTransferAware
+import sun.net.ftp.FtpClient
+
+import java.nio.channels.SeekableByteChannel
+import java.nio.file.*
+import java.nio.file.attribute.BasicFileAttributes
+import java.nio.file.attribute.FileAttribute
+import java.nio.file.attribute.FileAttributeView
+import java.nio.file.spi.FileSystemProvider
+
+@Slf4j
+@CompileStatic
+class WOWFileSystemProvider extends FileSystemProvider implements FileSystemTransferAware {
+
+ static final WOWFileSystemProvider INSTANCE = new WOWFileSystemProvider()
+
+ protected SchedulerClient schedulerClient = null
+
+ private transient final Object createSymlinkHelper = new Object()
+
+ void registerSchedulerClient(SchedulerClient schedulerClient) throws UnsupportedOperationException {
+ if (this.schedulerClient != null) {
+ throw new UnsupportedOperationException("WOW file system does not support multiple scheduler clients")
+ }
+ this.schedulerClient = schedulerClient
+ }
+
+ private FtpClient getConnection(final String node, String daemon ){
+ int trial = 0
+ while ( true ) {
+ try {
+ FtpClient ftpClient = FtpClient.create(daemon)
+ ftpClient.login("root", "password".toCharArray() )
+ ftpClient.enablePassiveMode( true )
+ ftpClient.setBinaryType()
+ return ftpClient
+ } catch ( IOException e ) {
+ if ( trial > 5 ) {
+ log.error("Cannot create FTP client: $daemon on $node", e)
+ throw e
+ }
+ sleep(Math.pow(2, trial++) as long)
+ daemon = schedulerClient.getDaemonOnNode(node)
+ }
+ }
+ }
+
+ @PackageScope Map getLocation(LocalPath path ){
+ String absolutePath = path.toAbsolutePath().toString()
+ Map response = schedulerClient.getFileLocation( absolutePath )
+ synchronized ( createSymlinkHelper ) {
+ if ( !path.createdSymlinks ) {
+ for (Map link : (response.symlinks as List