FinalizerThread for copyTasks

Signed-off-by: Lehmann_Fabian <[email protected]>

Author: Lehmann-Fabian

src/main/java/cws/k8s/scheduler/scheduler/FinalizerThread.java

@@ -0,0 +1,43 @@
+package cws.k8s.scheduler.scheduler;
+
+import lombok.RequiredArgsConstructor;
+
+import java.util.LinkedList;
+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;
+
+    @Override
+    public void run() {
+        while ( !Thread.currentThread().isInterrupted() ) {
+            synchronized ( finalizeItem ) {
+                while ( !items.isEmpty() ) {
+                    try {
+                        finalizeItem.accept( items.poll() );
+                    } catch ( Exception e ) {
+                        System.err.println( "Error finalizing item: " + e.getMessage() );
+                    }
+                }
+                try {
+                    finalizeItem.wait();
+                } catch ( InterruptedException e ) {
+                    Thread.currentThread().interrupt();
+                    System.err.println( "Finalizer thread interrupted: " + e.getMessage() );
+                }
+            }
+        }
+    }
+
+    public void addItem( T item ) {
+        synchronized ( finalizeItem ) {
+            items.add( item );
+            finalizeItem.notify();
+        }
+    }
+
+}

src/main/java/cws/k8s/scheduler/scheduler/LocationAwareSchedulerV2.java

@@ -67,6 +67,8 @@ public class LocationAwareSchedulerV2 extends SchedulerWithDaemonSet {
     protected final int copySameTaskInParallel;
     protected final int maxHeldCopyTaskReady;
 
+    private final FinalizerThread<Tuple<CopyTask, Boolean>> finalizerThread;
+
     /**
      * This value must be between 1 and 100.
      * 100 means that the data will be copied with full speed.
@@ -127,6 +129,8 @@ public LocationAwareSchedulerV2(
         this.copyInAdvance = new CopyInAdvanceNodeWithMostData( getCurrentlyCopying(), inputAlignment, this.copySameTaskInParallel );
         this.maxHeldCopyTaskReady = config.maxHeldCopyTaskReady == null ? 3 : config.maxHeldCopyTaskReady;
         this.prioPhaseThree = config.prioPhaseThree == null ? 70 : config.prioPhaseThree;
+        finalizerThread = new FinalizerThread<>( x -> processCopyTaskFinished( x.getA(), x.getB() ) );
+        finalizerThread.start();
     }
 
     @Override
@@ -283,15 +287,19 @@ private void undoReserveCopyTask( CopyTask copyTask ) {
     }
 
     public void copyTaskFinished( CopyTask copyTask, boolean success ) {
+        finalizerThread.addItem( new Tuple<CopyTask, Boolean>(copyTask, success) );
+    }
+
+    private void processCopyTaskFinished( CopyTask copyTask, boolean success ) {
         synchronized ( copyLock ) {
             freeLocations( copyTask.getAllLocationWrapper() );
             if( success ){
-                    copyTask.getInputFiles().parallelStream().forEach( TaskInputFileLocationWrapper::success );
-                    removeFromCopyingToNode( copyTask.getTask(), copyTask.getNodeLocation(), copyTask.getFilesForCurrentNode() );
-                    copyTask.getTask().preparedOnNode( copyTask.getNodeLocation() );
+                copyTask.getInputFiles().parallelStream().forEach( TaskInputFileLocationWrapper::success );
+                removeFromCopyingToNode( copyTask.getTask(), copyTask.getNodeLocation(), copyTask.getFilesForCurrentNode() );
+                copyTask.getTask().preparedOnNode( copyTask.getNodeLocation() );
             } else {
-                    removeFromCopyingToNode( copyTask.getTask(), copyTask.getNodeLocation(), copyTask.getFilesForCurrentNode() );
-                    handleProblematicCopy( copyTask );
+                removeFromCopyingToNode( copyTask.getTask(), copyTask.getNodeLocation(), copyTask.getFilesForCurrentNode() );
+                handleProblematicCopy( copyTask );
             }
         }
     }