diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java b/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java index 5e338ab..f321465 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java +++ b/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> 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); - this.starterFuture.cancel(true); - this.terminatorFuture.cancel(true); - } finally { - semaphore.release(); - } + stages.forEach(Stage::close); + this.starterFuture.cancel(true); + this.terminatorFuture.cancel(true); } 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> stages) { this.stages = stages; } + + public long getStartTime() { + return startTime; + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java index 4f5adf0..19aaa4e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java +++ b/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 idEnricherQueueFactory; private final StageQueueFactory nameEnricherQueueFactory; private final StageQueueFactory sendSignalQueueFactory; - private final UserPackageMapper userPackageMapper; private final FidIdEnricher fidIdEnricherHandler; private final FidNameEnricher fidNameEnricher; private final Supplier> pipelineTerminatorFactory; private final Function> pipelineStarterFactory; private final CreditTrackerSettings creditTrackerSettings; - private final Semaphore semaphore; private final PipelineBackoff pipelineBackoff; - private final MonitoringErrorProcesingService monitoringErrorProcesingService; public PipelinePackageManager(SignalRestService signalRestService, @Qualifier("idEnricherQueueFactory") StageQueueFactory 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()) { - 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()) + return 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 diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java new file mode 100644 index 0000000..d3962ac --- /dev/null +++ b/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 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); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java index 749b564..cd03e2f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java +++ b/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 lastQueue; private IPipeLineStarter pipeLineStarter; private IPipeLineTerminator pipeLineTerminator; - private Semaphore semaphore; private PipelineBackoff backoff; private PipeLineBuilderV2() { @@ -57,11 +55,6 @@ public class PipeLineBuilderV2 semaphore(Semaphore semaphore) { - this.semaphore = semaphore; - return this; - } - public PipeLineBuilderV2 backoff(PipelineBackoff backoff) { this.backoff = backoff; return this; @@ -69,7 +62,6 @@ public class PipeLineBuilderV2 { stages.forEach(stage -> { stage.setPipeline(p); @@ -78,8 +70,8 @@ public class PipeLineBuilderV2