pipeline orchestrator

src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java

@@ -3,7 +3,6 @@ package ru.nbch.credit_tracker.service.task;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
-import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -15,15 +14,16 @@ public class Pipeline {
     private Runnable starter;
     private IPipeLineTerminator<?,?> terminator;
     private List<Stage<?, ?>> stages;
-    private Semaphore semaphore;
     private Future<?> starterFuture;
     private Future<?> terminatorFuture;
     private final AtomicBoolean closed = new AtomicBoolean(false);
+    private volatile long startTime;
 
     public Pipeline() {
     }
 
     public void start(ExecutorService executorService) {
+        this.startTime = System.currentTimeMillis();
         this.starterFuture = executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), starter));
         this.terminatorFuture = executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), terminator));
     }
@@ -31,22 +31,14 @@ public class Pipeline {
     public void close() {
         if (closed.compareAndSet(false, true)) {
             log.info("closing the pipeline...");
-            try {
+            stages.forEach(Stage::close);
-                stages.forEach(Stage::close);
+            this.starterFuture.cancel(true);
-                this.starterFuture.cancel(true);
+            this.terminatorFuture.cancel(true);
-                this.terminatorFuture.cancel(true);
-            } finally {
-                semaphore.release();
-            }
         } else {
             log.info("pipeline already closed.");
         }
     }
 
-    public void setSemaphore(Semaphore semaphore) {
-        this.semaphore = semaphore;
-    }
-
     public Stage<?, ?> getLast() {
         return stages.getLast();
     }
@@ -74,4 +66,8 @@ public class Pipeline {
     public void setStages(List<Stage<?, ?>> stages) {
         this.stages = stages;
     }
+
+    public long getStartTime() {
+        return startTime;
+    }
 }

src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java

@@ -1,8 +1,5 @@
 package ru.nbch.credit_tracker.service.task;
 
-import java.time.LocalDateTime;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
@@ -21,29 +18,23 @@ import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher;
 import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher;
 import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
 import ru.nbch.credit_tracker.service.task.impl.Saver;
-import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
 import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilderV2;
-import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
 @Component
 @Slf4j
 public class PipelinePackageManager implements InitializingBean {
-    private final SignalRestService signalRestService;
     private final FlagsEnricher flagsEnricherV2;
     private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
     private final StageQueueFactory<SubjectPayload> nameEnricherQueueFactory;
     private final StageQueueFactory<PartFilePayload> sendSignalQueueFactory;
-    private final UserPackageMapper userPackageMapper;
     private final FidIdEnricher fidIdEnricherHandler;
     private final FidNameEnricher fidNameEnricher;
     private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
     private final Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
     private final CreditTrackerSettings creditTrackerSettings;
-    private final Semaphore semaphore;
     private final PipelineBackoff pipelineBackoff;
-    private final MonitoringErrorProcesingService monitoringErrorProcesingService;
 
     public PipelinePackageManager(SignalRestService signalRestService,
                                   @Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
@@ -57,32 +48,20 @@ public class PipelinePackageManager implements InitializingBean {
                                   CreditTrackerSettings creditTrackerSettings,
                                   PipelineBackoff pipelineBackoff, Saver saver,
                                   MonitoringErrorProcesingService monitoringErrorProcesingService) {
-        this.signalRestService = signalRestService;
         this.idEnricherQueueFactory = idEnricherQueueFactory;
         this.nameEnricherQueueFactory = nameEnricherQueueFactory;
         this.sendSignalQueueFactory = sendSignalQueueFactory;
-        this.userPackageMapper = userPackageMapper;
         this.fidIdEnricherHandler = fidIdEnricherHandler;
         this.pipelineTerminatorFactory = pipelineTerminatorFactory;
         this.pipelineStarterFactory = pipelineStarterFactory;
         this.flagsEnricherV2 = flagsEnricherV2;
         this.fidNameEnricher = fidNameEnricher;
         this.creditTrackerSettings = creditTrackerSettings;
-        this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests());
         this.pipelineBackoff = pipelineBackoff;
-        this.monitoringErrorProcesingService = monitoringErrorProcesingService;
     }
 
     public Pipeline buildAndRun(SetupParam setupParam) throws Exception {
-        if (!semaphore.tryAcquire()) {
+        return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
-            throw new RuntimeException("Too many concurrent requests");
-        }
-        log.info("throttler permits after acquire: {}", semaphore.availablePermits());
-
-        Long packageId = setupParam.packageId();
-        long startTime = System.currentTimeMillis();
-        try {
-            Pipeline pipeline = PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
                     .handler(fidNameEnricher)
                     .handlerResult(flagsEnricherV2)
                     .andThen()
@@ -91,45 +70,8 @@ public class PipelinePackageManager implements InitializingBean {
                     .andThen()
                 .fillInitialQueue(pipelineStarterFactory.apply(setupParam))
                 .pollFinishingQueue(pipelineTerminatorFactory.get())
-                .semaphore(semaphore)
                 .backoff(pipelineBackoff)
                 .build();
-
-            pipeline.getTerminator()
-                .doneFuture()
-                .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS)
-                .whenComplete((v, e) -> {
-                    try {
-                        long finishTime = System.currentTimeMillis();
-                        log.info("pipeline time passed: {} ms", finishTime - startTime);
-                        if (e instanceof XmlMergeCoordinator.PipelineTerminatorException) {
-                            log.info("when complete: terminator was interrupted.");
-                            return;
-                        } else if (e != null) {
-                            log.error(ExceptionUtils.getStackTrace(e));
-                            return;
-                        }
-                        userPackageMapper.setLastSearchedDt(packageId);
-                        SignalXmlFilePayload res = (SignalXmlFilePayload) v;
-                        log.trace("Signal file by path: {} is completed", res.signalXmlFilePath());
-                        boolean success = signalRestService.setupMonitoring(res.signalXmlFilePath());
-                        if (success) {
-                            userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now());
-                            monitoringErrorProcesingService.findErrors(packageId);
-                        }
-                    } catch (Exception ex) {
-                        log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex));
-                    } finally {
-                        log.info("packageId={}: pipeline finished", packageId);
-                        pipeline.close();
-                        log.info("semaphore amount: {}", semaphore.availablePermits());
-                    }
-                });
-            return pipeline;
-        } catch (Exception e) {
-            semaphore.release();
-            throw e;
-        }
     }
 
     @Override

src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java

@@ -0,0 +1,141 @@
+package ru.nbch.credit_tracker.service.task;
+
+import java.time.LocalDateTime;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
+import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
+import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService;
+import ru.nbch.credit_tracker.service.signal.SignalRestService;
+import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
+
+@Service
+public class PipelinePackageOrchestrator {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final CreditTrackerSettings creditTrackerSettings;
+    private final UserPackageMapper userPackageMapper;
+    private final SignalRestService signalRestService;
+    private final ExecutorService pipelineExecutorService;
+    private final ExecutorService saverExecutorService;
+    private final Object lock = new Object();
+    private final java.util.Deque<Launch> pending = new java.util.ArrayDeque<>();
+    private final MonitoringErrorProcesingService monitoringErrorProcesingService;
+    private final Semaphore semaphore;
+
+    public PipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings,
+                                       UserPackageMapper userPackageMapper,
+                                       SignalRestService signalRestService,
+                                       ExecutorService pipelineExecutorService,
+                                       ExecutorService saverExecutorService, MonitoringErrorProcesingService monitoringErrorProcesingService) {
+        this.creditTrackerSettings = creditTrackerSettings;
+        this.userPackageMapper = userPackageMapper;
+        this.signalRestService = signalRestService;
+        this.pipelineExecutorService = pipelineExecutorService;
+        this.saverExecutorService = saverExecutorService;
+        this.monitoringErrorProcesingService = monitoringErrorProcesingService;
+        this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests());
+    }
+
+    private record Launch(Runnable before, Pipeline pipeline, Long packageId) {
+    }
+
+    public void startOrDelay(Pipeline pipeline, Long packageId) {
+        startOrDelay(null, pipeline, packageId);
+    }
+
+    public void startOrDelay(Runnable before, Pipeline pipeline, Long packageId) {
+        synchronized (lock) {
+            if (!semaphore.tryAcquire()) {
+                log.debug("couldn't acquire permit from semaphore. delaying task for {}...", packageId);
+                pending.addLast(new Launch(before, pipeline, packageId));
+                return;
+            }
+            log.info("throttler permits after acquire: {}", semaphore.availablePermits());
+        }
+        innerStartOrDelay(before, pipeline, packageId);
+    }
+
+    public void innerStartOrDelay(Runnable before, Pipeline pipeline, Long packageId) {
+        try {
+            if (before != null) {
+                CompletableFuture
+                    .runAsync(before, saverExecutorService)
+                    .handle((v, t) -> {
+                        if (t != null) {
+                            log.error("`before` failed: {}", ExceptionUtils.getStackTrace(t));
+                            pipeline.close();
+                            checkForDelayedTasks(); // передать слот дальше или освободить
+                        } else {
+                            pipeline.start(pipelineExecutorService);
+                            addCallback(pipeline, packageId);
+                        }
+                        return null;
+                    });
+            } else {
+                pipeline.start(pipelineExecutorService);
+                addCallback(pipeline, packageId);
+            }
+        } catch (Exception e) {
+            log.error(ExceptionUtils.getStackTrace(e));
+            pipeline.close();
+            checkForDelayedTasks();
+        }
+    }
+
+    public void addCallback(Pipeline pipeline, Long packageId) {
+        pipeline.getTerminator()
+            .doneFuture()
+            .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS)
+            .whenComplete((v, e) -> {
+                try {
+                    long finishTime = System.currentTimeMillis();
+                    log.info("pipeline time passed: {} ms", finishTime - pipeline.getStartTime());
+                    if (e instanceof XmlMergeCoordinator.PipelineTerminatorException) {
+                        log.info("when complete: terminator was interrupted.");
+                        return;
+                    } else if (e != null) {
+                        log.error(ExceptionUtils.getStackTrace(e));
+                        return;
+                    }
+                    userPackageMapper.setLastSearchedDt(packageId);
+                    SignalXmlFilePayload res = (SignalXmlFilePayload) v;
+                    log.trace("Signal file by path: {} is completed", res.signalXmlFilePath());
+                    boolean success = signalRestService.setupMonitoring(res.signalXmlFilePath());
+                    if (success) {
+                        userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now());
+                        monitoringErrorProcesingService.findErrors(packageId);
+                    }
+                } catch (Exception ex) {
+                    log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex));
+                } finally {
+                    log.info("packageId={}: pipeline finished", packageId);
+                    pipeline.close();
+                    checkForDelayedTasks();
+                    log.info("semaphore amount: {}", semaphore.availablePermits());
+                }
+            });
+    }
+
+    private void checkForDelayedTasks() {
+        log.debug("checking for delayed pipelines...");
+        Launch delayedLaunch;
+        synchronized (lock) {
+            delayedLaunch = pending.pollFirst();
+            if (delayedLaunch == null) {
+                log.debug("no delayed pipelines found; releasing semaphore...");
+                semaphore.release();
+                return;
+            }
+        }
+
+        log.info("delayed pipeline found: packageId={}. continue work...", delayedLaunch.packageId);
+        innerStartOrDelay(delayedLaunch.before, delayedLaunch.pipeline, delayedLaunch.packageId);
+    }
+}

src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java

@@ -4,7 +4,6 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Semaphore;
 import ru.nbch.credit_tracker.model.StagePayload;
 import ru.nbch.credit_tracker.service.task.Pipeline;
 import ru.nbch.credit_tracker.service.task.PipelineBackoff;
@@ -17,7 +16,6 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
     private BlockingQueue<LAST> lastQueue;
     private IPipeLineStarter<FIRST> pipeLineStarter;
     private IPipeLineTerminator<LAST, ?> pipeLineTerminator;
-    private Semaphore semaphore;
     private PipelineBackoff backoff;
 
     private PipeLineBuilderV2() {
@@ -57,11 +55,6 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
         return this;
     }
 
-    public PipeLineBuilderV2<FIRST, LAST> semaphore(Semaphore semaphore) {
-        this.semaphore = semaphore;
-        return this;
-    }
-
     public PipeLineBuilderV2<FIRST, LAST> backoff(PipelineBackoff backoff) {
         this.backoff = backoff;
         return this;
@@ -69,7 +62,6 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
 
     public Pipeline build() {
         Pipeline p = new Pipeline();
-        p.setSemaphore(semaphore);
         p.setStarter(() -> {
             stages.forEach(stage -> {
                 stage.setPipeline(p);
@@ -78,8 +70,8 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
             });
             pipeLineStarter.startFillingQueue(firstStage.getIn());
         });
-            pipeLineTerminator.setPollingQueue(lastQueue);
+        pipeLineTerminator.setPollingQueue(lastQueue);
-            p.setTerminator(pipeLineTerminator);
+        p.setTerminator(pipeLineTerminator);
         p.setStages(stages);
         return p;
     }