Merge pull request #7 from ftschirpke/wow_as_fs

Support for WOW scheduler

Author: Lehmann-Fabian

.github/workflows/build.yml

@@ -29,6 +29,14 @@ jobs:
           fetch-depth: 1
           submodules: true
 
+      - name: Install dependencies for C (cross-)compilation
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y clang lld gcc-aarch64-linux-gnu libc6-dev-arm64-cross
+
+      - name: Compile C script using make
+        run: make c_scripts
+
       - name: Setup Java ${{ matrix.java_version }}
         uses: actions/setup-java@v3
         with:

.gitignore

@@ -10,3 +10,4 @@ out
 /gradle.properties
 /logs.tmp
 *.jar
+/plugins/nf-cws/src/resources/nf-cws/getStatsAndResolveSymlinks_*

Makefile

@@ -7,18 +7,52 @@ else
 mm = 
 endif 
 
+C_SCRIPT_SRC = plugins/nf-cws/src/resources/nf-cws/getStatsAndResolveSymlinks.c
+C_SCRIPT_AARCH64 = plugins/nf-cws/src/resources/nf-cws/getStatsAndResolveSymlinks_linux_aarch64
+C_SCRIPT_X86_64 = plugins/nf-cws/src/resources/nf-cws/getStatsAndResolveSymlinks_linux_x86_64
+C_SCRIPT_ALL_TARGETS = $(C_SCRIPT_X86_64) $(C_SCRIPT_AARCH64)
+
+arch:=$(shell uname -m)
+
+ifneq ($(arch),x86_64)
+$(info ====================================================================================)
+$(info This Makefile assumes to be run on an x86_64 build system. Found: $(arch))
+$(info As an alternative you can adjust the Makefile or build $(C_SCRIPT_SRC) yourself for the following targets:)
+$(info - target aarch64-linux-gnu: saved to $(C_SCRIPT_AARCH64))
+$(info - target x86_64-linux-gnu: saved to $(C_SCRIPT_X86_64))
+$(error Aborting)
+endif
+
+c_scripts: $(C_SCRIPT_ALL_TARGETS)
+
+$(C_SCRIPT_AARCH64): $(C_SRIPT_SRC)
+	clang -static \
+		-target aarch64-linux-gnu \
+		--sysroot=/usr/aarch64-linux-gnu \
+		plugins/nf-cws/src/resources/nf-cws/getStatsAndResolveSymlinks.c \
+		-fuse-ld=lld \
+		-o $@
+
+
+$(C_SCRIPT_X86_64): $(C_SCRIPT_SRC)
+	clang -static \
+		-target x86_64-linux-gnu \
+		plugins/nf-cws/src/resources/nf-cws/getStatsAndResolveSymlinks.c \
+		-fuse-ld=lld \
+		-o $@
+
 clean:
 	rm -rf .nextflow*
 	rm -rf work
 	rm -rf build
 	rm -rf plugins/*/build
+	rm -f $(C_SCRIPT_ALL_TARGETS)
 	./gradlew clean
 
 compile:
 	./gradlew :nextflow:exportClasspath compileGroovy
 	@echo "DONE `date`"
 
-
 check:
 	./gradlew check
 
@@ -55,7 +89,7 @@ assemble:
 # generate build zips under build/plugins
 # you can install the plugin copying manually these files to $HOME/.nextflow/plugins
 #
-buildPlugins:
+buildPlugins: $(C_SCRIPT_ALL_TARGETS)
 	./gradlew copyPluginZip
 
 #
@@ -69,4 +103,4 @@ upload-plugins:
 	./gradlew plugins:upload
 
 publish-index:
-	./gradlew plugins:publishIndex
+	./gradlew plugins:publishIndex

README.md

@@ -17,6 +17,7 @@ java {
     toolchain {
         languageVersion = JavaLanguageVersion.of(21)
     }
+
     sourceCompatibility = 17
     targetCompatibility = 17
 }

buildSrc/src/main/groovy/io.nextflow.groovy-common-conventions.gradle

@@ -34,7 +34,7 @@ sourceSets {
 }
 
 ext{
-    nextflowVersion = '24.11.0-edge'
+    nextflowVersion = '25.01.0-edge'
 }
 
 dependencies {

plugins/nf-cws/build.gradle

@@ -16,6 +16,14 @@ class CWSConfig {
 
     String getStrategy() { target.strategy as String ?: 'FIFO' }
 
+    boolean strategyIsLocationAware() {
+        switch(target.strategy) {
+            case "wow": return true
+            default:
+                return false
+        }
+    }
+
     String getCostFunction() { target.costFunction as String }
 
     String getMemoryPredictor() { target.memoryPredictor as String }
@@ -30,6 +38,10 @@ class CWSConfig {
         return s ? new MemoryUnit(s) : null
     }
 
+    Integer getMaxCopyTasksPerNode() { target.maxCopyTasksPerNode as Integer }
+
+    Integer getMaxWaitingCopyTasksPerNode() { target.maxWaitingCopyTasksPerNode as Integer }
+
     int getBatchSize() {
         String s = target.batchSize as String
         //Default: 1 -> No batching

plugins/nf-cws/src/main/nextflow/cws/CWSConfig.groovy

@@ -1,8 +1,15 @@
 package nextflow.cws
 
 import groovy.transform.CompileStatic
+import nextflow.cws.wow.file.LocalPath
+import nextflow.cws.wow.file.OfflineLocalPath
+import nextflow.cws.wow.file.WorkdirPath
+import nextflow.cws.wow.filesystem.WOWFileSystemProvider
+import nextflow.cws.wow.serializer.LocalPathSerializer
+import nextflow.file.FileHelper
 import nextflow.plugin.BasePlugin
 import nextflow.trace.TraceRecord
+import nextflow.util.KryoHelper
 import org.pf4j.PluginWrapper
 
 @CompileStatic
@@ -14,6 +21,10 @@ class CWSPlugin extends BasePlugin {
 
     private static void registerTraceFields() {
         TraceRecord.FIELDS.putAll( [
+                infiles_time:                          'num',
+                outfiles_time:                         'num',
+                create_bash_wrapper_time:              'num',
+                create_request_time:                   'num',
                 submit_to_scheduler_time:              'num',
                 submit_to_k8s_time:                    'num',
                 scheduler_time_in_queue:               'num',
@@ -33,13 +44,29 @@ class CWSPlugin extends BasePlugin {
                 scheduler_delta_submitted_batch_end:   'num',
                 memory_adapted:                        'mem',
                 input_size:                            'num',
+                scheduler_files_bytes:                 'num',
+                scheduler_files_node_bytes:            'num',
+                scheduler_files_node_other_task_bytes: 'num',
+                scheduler_files:                       'num',
+                scheduler_files_node:                  'num',
+                scheduler_files_node_other_task:       'num',
+                scheduler_depending_task:              'num',
+                scheduler_location_count:              'num',
+                scheduler_nodes_to_copy_from:          'num',
+                scheduler_no_alignment_found:          'num',
+                scheduler_time_delta_phase_three:      'str',
+                scheduler_copy_tasks:                  'num',
         ] )
     }
 
     @Override
     void start() {
         super.start()
         registerTraceFields()
+        KryoHelper.register( LocalPath, LocalPathSerializer )
+        KryoHelper.register( OfflineLocalPath, LocalPathSerializer )
+        KryoHelper.register( WorkdirPath, LocalPathSerializer )
+        FileHelper.getOrInstallProvider(WOWFileSystemProvider)
     }
 
 }

plugins/nf-cws/src/main/nextflow/cws/CWSPlugin.groovy

@@ -1,7 +1,9 @@
 package nextflow.cws
 
+import groovy.transform.CompileStatic
 import nextflow.cws.processor.SchedulerBatch
 
+@CompileStatic
 class CWSSchedulerBatch extends SchedulerBatch {
 
     private SchedulerClient schedulerClient

plugins/nf-cws/src/main/nextflow/cws/CWSSchedulerBatch.groovy

@@ -1,8 +1,11 @@
 package nextflow.cws
 
+import groovy.transform.CompileStatic
+
 /**
  * Central, global instance to manage all generated Executor instances
  */
+@CompileStatic
 class CWSSession {
 
     static final CWSSession INSTANCE = new CWSSession()