Improve Thread safety

Lehmann-Fabian · Lehmann-Fabian · commit d1900cf4fc17 · 2025-06-01T21:36:25.000+02:00
Signed-off-by: Lehmann_Fabian &lt;fabian.lehmann@informatik.hu-berlin.de&gt;
diff --git a/src/main/java/cws/k8s/scheduler/scheduler/FinalizerThread.java b/src/main/java/cws/k8s/scheduler/scheduler/FinalizerThread.java
@@ -6,28 +6,44 @@
 import java.util.Queue;
 import java.util.function.Consumer;
 
-@RequiredArgsConstructor
 public class FinalizerThread<T> extends Thread {
 
     private final Queue<T> items = new LinkedList<>();
     private final Consumer <? super T> finalizeItem;
 
+    public FinalizerThread(Consumer <? super T> finalizeItem) {
+        super( "FinalizerThread" );
+        this.finalizeItem = finalizeItem;
+    }
+
     @Override
     public void run() {
         while ( !Thread.currentThread().isInterrupted() ) {
+            Queue<T> toProcess = new LinkedList<>();
             synchronized ( finalizeItem ) {
                 while ( !items.isEmpty() ) {
                     try {
-                        finalizeItem.accept( items.poll() );
+                        toProcess.add( items.poll() );
                     } catch ( Exception e ) {
                         System.err.println( "Error finalizing item: " + e.getMessage() );
                     }
                 }
+            }
+            for ( T item : toProcess ) {
                 try {
-                    finalizeItem.wait();
-                } catch ( InterruptedException e ) {
-                    Thread.currentThread().interrupt();
-                    System.err.println( "Finalizer thread interrupted: " + e.getMessage() );
+                    finalizeItem.accept( item );
+                } catch ( Exception e ) {
+                    System.err.println( "Error finalizing item: " + e.getMessage() );
+                }
+            }
+            synchronized ( finalizeItem ) {
+                if ( items.isEmpty() ) {
+                    try {
+                        finalizeItem.wait();
+                    } catch ( InterruptedException e ) {
+                        Thread.currentThread().interrupt();
+                        System.err.println( "Finalizer thread interrupted: " + e.getMessage() );
+                    }
                 }
             }
         }
@@ -36,7 +52,7 @@ public void run() {
     public void addItem( T item ) {
         synchronized ( finalizeItem ) {
             items.add( item );
-            finalizeItem.notify();
+            finalizeItem.notifyAll();
         }
     }
 
diff --git a/src/main/java/cws/k8s/scheduler/scheduler/LocationAwareSchedulerV2.java b/src/main/java/cws/k8s/scheduler/scheduler/LocationAwareSchedulerV2.java
@@ -302,6 +302,7 @@ private void processCopyTaskFinished( CopyTask copyTask, boolean success ) {
                 handleProblematicCopy( copyTask );
             }
         }
+        informResourceChange();
     }
 
     private void handleProblematicCopy( CopyTask copyTask ){
diff --git a/src/main/java/cws/k8s/scheduler/scheduler/Scheduler.java b/src/main/java/cws/k8s/scheduler/scheduler/Scheduler.java
@@ -69,10 +69,10 @@ public abstract class Scheduler implements Informable {
 
         PodHandler handler = new PodHandler(this );
 
-        schedulingThread = new TaskprocessingThread( unscheduledTasks, this::schedule );
+        schedulingThread = new TaskprocessingThread( "Scheduling Thread", unscheduledTasks, this::schedule );
         schedulingThread.start();
 
-        finishThread = new TaskprocessingThread(unfinishedTasks, this::terminateTasks );
+        finishThread = new TaskprocessingThread( "Finisher Thread", unfinishedTasks, this::terminateTasks );
         finishThread.start();
 
         log.info("Start watching: {}", this.namespace );
diff --git a/src/main/java/cws/k8s/scheduler/scheduler/SchedulerWithDaemonSet.java b/src/main/java/cws/k8s/scheduler/scheduler/SchedulerWithDaemonSet.java
@@ -114,32 +114,35 @@ int terminateTasks(List<Task> finishedTasks) {
                     log.info( "Pod finished with exitCode: {}", exitCode );
                     //Init failure
                     final Path workdir = Paths.get(finishedTask.getWorkingDir());
-                    if ( exitCode == 123 && Files.exists( workdir.resolve(".command.init.failure") ) ) {
-                        log.info( "Task {} ({}) had an init failure: won't parse the in- and out files", finishedTask.getConfig().getRunName(), finishedTask.getConfig().getName() );
-                    } else {
-                        final Set<OutputFile> newAndUpdatedFiles = taskResultParser.getNewAndUpdatedFiles(
-                                workdir,
-                                finishedTask.getNode().getNodeLocation(),
-                                !finishedTask.wasSuccessfullyExecuted(),
-                                finishedTask
-                        );
-                        for (OutputFile newAndUpdatedFile : newAndUpdatedFiles) {
-                            if( newAndUpdatedFile instanceof PathLocationWrapperPair pathLocationWrapperPair ) {
-                                hierarchyWrapper.addFile(
-                                        newAndUpdatedFile.getPath(),
-                                        pathLocationWrapperPair.getLocationWrapper()
-                                );
-                            } else if ( newAndUpdatedFile instanceof SymlinkOutput symlinkOutput ){
-                                hierarchyWrapper.addSymlink( newAndUpdatedFile.getPath(), symlinkOutput.getDst() );
+                    final Set<OutputFile> newAndUpdatedFiles = taskResultParser.getNewAndUpdatedFiles(
+                            workdir,
+                            finishedTask.getNode().getNodeLocation(),
+                            !finishedTask.wasSuccessfullyExecuted(),
+                            finishedTask
+                    );
+                    final Set<PathLocationWrapperPair> outputFiles = new HashSet<>();
+                    for (OutputFile newAndUpdatedFile : newAndUpdatedFiles) {
+                        if( newAndUpdatedFile instanceof PathLocationWrapperPair pathLocationWrapperPair ) {
+                            final LocationWrapper locationWrapper = hierarchyWrapper.addFile(
+                                    newAndUpdatedFile.getPath(),
+                                    pathLocationWrapperPair.getLocationWrapper()
+                            );
+                            if ( locationWrapper != null ) {
+                                outputFiles.add( pathLocationWrapperPair );
                             }
+                        } else if ( newAndUpdatedFile instanceof SymlinkOutput symlinkOutput ){
+                            hierarchyWrapper.addSymlink( newAndUpdatedFile.getPath(), symlinkOutput.getDst() );
                         }
                     }
                 }
             } catch ( Exception e ){
                 log.info( "Problem while finishing task: {} ({})", finishedTask.getConfig().getRunName(), finishedTask.getConfig().getName(), e );
             }
-            super.taskWasFinished( finishedTask );
         });
+        // run in a separate thread to avoid blocking the finishedTasks list
+        for ( Task finishedTask : finishedTasks ) {
+            super.taskWasFinished( finishedTask );
+        }
         return 0;
     }
 
diff --git a/src/main/java/cws/k8s/scheduler/scheduler/TaskprocessingThread.java b/src/main/java/cws/k8s/scheduler/scheduler/TaskprocessingThread.java
@@ -9,14 +9,19 @@
 import java.util.function.Function;
 
 @Slf4j
-@RequiredArgsConstructor
 public class TaskprocessingThread extends Thread {
 
     private final List<Task> unprocessedTasks;
     private final Function<List<Task>, Integer> function;
 
     private boolean otherResourceChange = false;
 
+    public TaskprocessingThread( String name, List<Task> unprocessedTasks, Function<List<Task>, Integer> function) {
+        super(name );
+        this.unprocessedTasks = unprocessedTasks;
+        this.function = function;
+    }
+
     public void otherResourceChange() {
         otherResourceChange = true;
     }
diff --git a/src/main/java/cws/k8s/scheduler/scheduler/la2/copystrategy/LaListener.java b/src/main/java/cws/k8s/scheduler/scheduler/la2/copystrategy/LaListener.java
@@ -56,7 +56,6 @@ public void onClose( int exitCode, String reason ) {
             log.error( "Copy task was not finished, but closed. ExitCode: " + exitCode + " Reason: " + reason );
             scheduler.copyTaskFinished( copyTask, exitCode == 0 );
         }
-        scheduler.informResourceChange();
     }
 
     @Override

Original file line number	Diff line number	Diff line change
`@@ -302,6 +302,7 @@ private void processCopyTaskFinished( CopyTask copyTask, boolean success ) {`
`302`	`302`	`handleProblematicCopy( copyTask );`
`303`	`303`	`}`
`304`	`304`	`}`
	`305`	`+ informResourceChange();`
`305`	`306`	`}`
`306`	`307`
`307`	`308`	`private void handleProblematicCopy( CopyTask copyTask ){`
Original file line number	Diff line number	Diff line change
`@@ -56,7 +56,6 @@ public void onClose( int exitCode, String reason ) {`
`56`	`56`	`log.error( "Copy task was not finished, but closed. ExitCode: " + exitCode + " Reason: " + reason );`
`57`	`57`	`scheduler.copyTaskFinished( copyTask, exitCode == 0 );`
`58`	`58`	`}`
`59`		`- scheduler.informResourceChange();`
`60`	`59`	`}`
`61`	`60`
`62`	`61`	`@Override`