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 9b1c135..5e338ab 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 @@ -1,16 +1,52 @@ 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; import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; +import ru.nbch.credit_tracker.utils.LogUtil; public class Pipeline { + private final Logger log = LoggerFactory.getLogger(getClass()); 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); public Pipeline() { } + public void start(ExecutorService executorService) { + this.starterFuture = executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), starter)); + this.terminatorFuture = executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), terminator)); + } + + 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(); + } + } else { + log.info("pipeline already closed."); + } + } + + public void setSemaphore(Semaphore semaphore) { + this.semaphore = semaphore; + } + public Stage getLast() { return stages.getLast(); } 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 927c833..2eae8fa 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 @@ -18,10 +18,10 @@ import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher; import ru.nbch.credit_tracker.service.task.impl.FlagsEnricherV2; import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler; +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.LogUtil; import ru.nbch.credit_tracker.utils.error.ExceptionUtils; @Component @@ -84,17 +84,19 @@ public class PipelinePackageManager { .andThen() .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath)) .pollFinishingQueue(pipelineTerminatorFactory.get()) + .semaphore(semaphore) .build(); - executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), pipeline.getStarter())); - executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), pipeline.getTerminator())); - + pipeline.start(executorService); pipeline.getTerminator() .doneFuture() .orTimeout(creditTrackerSettings.getTimeout(), TimeUnit.MINUTES) .whenComplete((v, e) -> { try { - if (e != null) { + if (e instanceof XmlMergeCoordinator.PipelineTerminatorException) { + log.info("when complete: terminator was interrupted."); + return; + } else if (e != null) { log.error(ExceptionUtils.getStackTrace(e)); return; } @@ -105,8 +107,8 @@ public class PipelinePackageManager { log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex)); } finally { log.info("packageId={}: pipeline finished", packageId); - pipeline.getStages().forEach(Stage::close); - semaphore.release(); + pipeline.close(); + log.info("semaphore amount: {}", semaphore.availablePermits()); } }); } catch (Exception e) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java index 97714a6..4422e4a 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java @@ -9,9 +9,11 @@ import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; +import org.mybatis.spring.MyBatisSystemException; import ru.nbch.credit_tracker.model.StagePayload; import ru.nbch.credit_tracker.utils.LogUtil; @@ -24,6 +26,7 @@ public class Stage implements Ru private final ExecutorService ioPool; private Consumer consumer = null; private final String dispatcherName; + private Pipeline pipeline; private static final ThreadPoolExecutor.CallerRunsPolicy callerRunsPolicy = new ThreadPoolExecutor.CallerRunsPolicy(); @@ -54,6 +57,10 @@ public class Stage implements Ru this.POISON = POISON; } + public void setPipeline(Pipeline pipeline) { + this.pipeline = pipeline; + } + public void startRunning() { ioPool.submit(this); } @@ -61,6 +68,7 @@ public class Stage implements Ru @Override public void run() { List> inflight = new ArrayList<>(); + log.info("this is main thread for {}", dispatcherName); try { int counter = 0; while (true) { @@ -74,33 +82,46 @@ public class Stage implements Ru // дождаться всех in-flight задач for (Future f : inflight) { if (wasInterrupted()) { - log.info("stage {} was interrupted while waiting for tasks to finish", - dispatcher.getClass().getSimpleName()); + log.info("stage {} was interrupted while waiting for tasks to finish", dispatcherName); break; } f.get(); } putQuiet(out, POISON); - log.info("stage {} finished.", dispatcher.getClass().getSimpleName()); + log.info("stage {} finished.", dispatcherName); return; } counter++; int finalCounter = counter; + AtomicInteger retryForSingleTask = new AtomicInteger(0); Future f = ioPool.submit(() -> { - try { - if (wasInterrupted()) { - log.info("stage {} task {} interrupted...", dispatcher.getClass().getSimpleName(), finalCounter); + while (retryForSingleTask.incrementAndGet() < 3) { + try { + if (Thread.currentThread().isInterrupted()) { + log.info("stage {} was interrupted. skipping task {}...", dispatcherName, finalCounter); + return null; + } + log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter); + Z p = dispatcher.handle(b); + if (Thread.currentThread().isInterrupted()) { + log.info("after {} task {} finished, worker has been interrupted. not putting result in queue.", + dispatcherName, finalCounter); + return null; + } + putQuiet(out, p); + log.info("stage {} task {} ended.", dispatcher.getClass().getSimpleName(), finalCounter); + return p; + } catch (MyBatisSystemException e) { + log.warn("{} RETRY AFTER ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e)); + } catch (Exception e) { + log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e)); + pipeline.close(); return null; } - log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter); - Z p = dispatcher.handle(b); - putQuiet(out, p); - log.info("stage {} task {} ended.", dispatcher.getClass().getSimpleName(), finalCounter); - return p; - } catch (Exception e) { - log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e)); - return null; } + log.error("tried for {} times, failed batch...", retryForSingleTask.get()); + pipeline.close(); + return null; } ); inflight.add(f); @@ -124,9 +145,9 @@ public class Stage implements Ru } private boolean wasInterrupted() { - if (Thread.currentThread().isInterrupted()) { + if (Thread.interrupted()) { log.warn("pipeline was interrupted..."); - putQuiet(out, POISON); + //putQuiet(out, POISON); return true; } return false; @@ -153,7 +174,7 @@ public class Stage implements Ru try { return q.poll(timeout.toMillis(), TimeUnit.MILLISECONDS); } catch (InterruptedException e) { - log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e)); + log.error("Main thread for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e)); Thread.currentThread().interrupt(); return null; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java index bc9cace..24e42b2 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java @@ -12,6 +12,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; +import ru.nbch.credit_tracker.exceptions.SignalClientException; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; @@ -35,6 +36,12 @@ public class KickPipeLineImpl implements IPipeLineStarter { this.BATCH_SIZE = creditTrackerSettings.getBatchSize(); } + private static class PipelineStarterException extends SignalClientException { + public PipelineStarterException() { + super(null, null, null); + } + } + @Override public void startFillingQueue(BlockingQueue firstQueue) { log.info("Start registering subjects. PackageId: {}", packageId); @@ -55,8 +62,8 @@ public class KickPipeLineImpl implements IPipeLineStarter { try { firstQueue.put(new SubjectPayload(packageId, subjectProfileBatch)); } catch (InterruptedException e) { - log.error("{}", ExceptionUtils.getStackTrace(e)); - throw new RuntimeException(e); + log.error("Pipeline starter was interrupted!"); + throw new PipelineStarterException(); } request.getSubjects().getSubject().clear(); } @@ -66,6 +73,8 @@ public class KickPipeLineImpl implements IPipeLineStarter { BATCH_SIZE, "/MonitoringRequest/Subjects/Subject"); } catch (IOException e) { log.error("{}", ExceptionUtils.getStackTrace(e)); + } catch (PipelineStarterException e) { + log.info("Pipeline starter shitting down..."); } finally { log.debug("Send poison pill"); try { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java index 6b495a1..6e6378b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java @@ -59,11 +59,16 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator inputs) throws Exception { 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 46725a3..cd13f58 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,6 +4,7 @@ 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.Stage; @@ -15,6 +16,7 @@ public class PipeLineBuilderV2 lastQueue; private IPipeLineStarter pipeLineStarter; private IPipeLineTerminator pipeLineTerminator; + private Semaphore semaphore; private PipeLineBuilderV2() { this.stages = new ArrayList<>(); @@ -53,10 +55,19 @@ public class PipeLineBuilderV2 semaphore(Semaphore semaphore) { + this.semaphore = semaphore; + return this; + } + public Pipeline build() { Pipeline p = new Pipeline(); + p.setSemaphore(semaphore); p.setStarter(() -> { - stages.forEach(Stage::startRunning); + stages.forEach(stage -> { + stage.setPipeline(p); + stage.startRunning(); + }); pipeLineStarter.startFillingQueue(firstStage.getIn()); }); pipeLineTerminator.setPollingQueue(lastQueue);