diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java index bf73de1..03dd774 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java @@ -21,6 +21,8 @@ public class CreditTrackerSettings { private DataSourceSettings scoringDataSourceSettings; @NestedConfigurationProperty private XsdValidation xsdValidation; + @NestedConfigurationProperty + private PipelineSettings pipeline; private String signalsRootUri; private String signalsToken; @@ -31,11 +33,6 @@ public class CreditTrackerSettings { private Long monitoringInterval; private Long updateFlagsInterval; private Long daysToUpdateFlags; - private String encoding; - - private Integer batchSize; - private Integer timeout; - private String dbVersion; } diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java new file mode 100644 index 0000000..f2279df --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java @@ -0,0 +1,18 @@ +package ru.nbch.credit_tracker.config.settings; + +import java.time.Duration; +import java.util.List; +import org.springframework.boot.context.properties.bind.DefaultValue; + +public record PipelineSettings( + @DefaultValue("5000") Integer subjectBatchSize, + @DefaultValue("15") Integer nameEnricherStageConcurrency, + @DefaultValue("10") Integer idEnricherStageConcurrency, + @DefaultValue("5") Integer errorAttemptCount, + @DefaultValue("2s") Duration errorAttemptDelay, + @DefaultValue("5m") Duration globalTimeout, + @DefaultValue("3") Integer parallelRequests, + @DefaultValue("org.springframework.dao.DataAccessException") + List> retryableExceptions +) { +} diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java index 2baaef7..73e0fd6 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java @@ -1,7 +1,7 @@ package ru.nbch.credit_tracker.config.task; +import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; import org.springframework.beans.factory.FactoryBean; import ru.nbch.credit_tracker.model.StagePayload; @@ -18,7 +18,7 @@ public class StageQueueFactory implements FactoryBean getObject() throws Exception { - return new LinkedBlockingQueue<>(capacity); + return new ArrayBlockingQueue<>(capacity); } @Override diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineBackoff.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineBackoff.java new file mode 100644 index 0000000..327f565 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineBackoff.java @@ -0,0 +1,33 @@ +package ru.nbch.credit_tracker.service.task; + +import java.time.Duration; +import java.util.List; +import lombok.Getter; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; + +@Getter +@Service +public class PipelineBackoff { + private final Integer retryCount; + private final Duration delay; + private final List> retryableExceptions; + + public PipelineBackoff(CreditTrackerSettings creditTrackerSettings) { + this.retryCount = creditTrackerSettings.getPipeline().errorAttemptCount(); + this.delay = creditTrackerSettings.getPipeline().errorAttemptDelay(); + this.retryableExceptions = creditTrackerSettings.getPipeline().retryableExceptions(); + } + + public void backoff() { + try { + Thread.sleep(delay.toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + public boolean isRetryable(final Throwable throwable) { + return retryableExceptions.stream().anyMatch(aClass -> aClass.isAssignableFrom(throwable.getClass())); + } +} 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 3e6e01d..1032b1d 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 @@ -17,7 +17,7 @@ import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; 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.FlagsEnricher; 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; @@ -29,7 +29,7 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils; @Slf4j public class PipelinePackageManager { private final SignalRestService signalRestService; - private final FlagsEnricherV2 flagsEnricherV2; + private final FlagsEnricher flagsEnricherV2; private final StageQueueFactory saveSubjectQueueFactory; private final StageQueueFactory idEnricherQueueFactory; private final StageQueueFactory sendSignalQueueFactory; @@ -44,6 +44,7 @@ public class PipelinePackageManager { return t; }); private final Semaphore semaphore; + private final PipelineBackoff pipelineBackoff; private final static int MAX_PARALLEL_REQUESTS = 3; public PipelinePackageManager(SignalRestService signalRestService, @@ -53,8 +54,9 @@ public class PipelinePackageManager { SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler, Supplier> pipelineTerminatorFactory, BiFunction> pipelineStarterFactory, - FlagsEnricherV2 flagsEnricherV2, - CreditTrackerSettings creditTrackerSettings) { + FlagsEnricher flagsEnricherV2, + CreditTrackerSettings creditTrackerSettings, + PipelineBackoff pipelineBackoff) { this.signalRestService = signalRestService; this.saveSubjectQueueFactory = saveSubjectQueueFactory; this.idEnricherQueueFactory = idEnricherQueueFactory; @@ -65,7 +67,8 @@ public class PipelinePackageManager { this.pipelineStarterFactory = pipelineStarterFactory; this.flagsEnricherV2 = flagsEnricherV2; this.creditTrackerSettings = creditTrackerSettings; - this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS); + this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests()); + this.pipelineBackoff = pipelineBackoff; } public void buildAndRun(Long packageId, Path pkgPath) throws Exception { @@ -84,12 +87,13 @@ public class PipelinePackageManager { .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath)) .pollFinishingQueue(pipelineTerminatorFactory.get()) .semaphore(semaphore) + .backoff(pipelineBackoff) .build(); pipeline.start(executorService); pipeline.getTerminator() .doneFuture() - .orTimeout(creditTrackerSettings.getTimeout(), TimeUnit.MINUTES) + .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS) .whenComplete((v, e) -> { try { if (e instanceof XmlMergeCoordinator.PipelineTerminatorException) { 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 19f6c97..264cc48 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 @@ -1,6 +1,5 @@ package ru.nbch.credit_tracker.service.task; -import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; @@ -13,8 +12,8 @@ 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.TaskTimer; import ru.nbch.credit_tracker.utils.LogUtil; @Slf4j @@ -26,6 +25,8 @@ public class Stage implements Ru private Consumer consumer = null; private final String dispatcherName; private Pipeline pipeline; + private PipelineBackoff pipelineBackoff; + private final TaskTimer stopWatch = new TaskTimer(); private static final ThreadPoolExecutor.CallerRunsPolicy callerRunsPolicy = new ThreadPoolExecutor.CallerRunsPolicy(); @@ -33,6 +34,9 @@ public class Stage implements Ru public Stage(Handler dispatcher, BlockingQueue in, BlockingQueue out) { + if (dispatcher.preferredConcurrency() < 2) { + throw new IllegalStateException("preferred concurrency must be at least 2"); + } this.dispatcher = dispatcher; this.in = in; this.out = out; @@ -54,14 +58,6 @@ public class Stage implements Ru ); } - public void setPipeline(Pipeline pipeline) { - this.pipeline = pipeline; - } - - public void startRunning() { - ioPool.submit(this); - } - @Override public void run() { List> inflight = new ArrayList<>(); @@ -69,11 +65,10 @@ public class Stage implements Ru try { int counter = 0; while (true) { - X b = takeQuiet(in, Duration.ofSeconds(5)); + X b = takeQuiet(); if (wasInterrupted()) { break; } - if (b == null) continue; if (b.isPoison()) { log.info("Got poison"); // дождаться всех in-flight задач @@ -84,6 +79,7 @@ public class Stage implements Ru } f.get(); } + log.trace("Stage time stats: {}", stopWatch.info()); putQuiet(out, dispatcher.cookPoison()); log.info("stage {} finished.", dispatcherName); return; @@ -92,14 +88,17 @@ public class Stage implements Ru int finalCounter = counter; AtomicInteger retryForSingleTask = new AtomicInteger(0); Future f = ioPool.submit(() -> { - while (retryForSingleTask.incrementAndGet() < 3) { + while (retryForSingleTask.incrementAndGet() < pipelineBackoff.getRetryCount()) { 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); + Z p; + try (var m = stopWatch.start(dispatcherName)){ + p = dispatcher.handle(b); + } if (Thread.currentThread().isInterrupted()) { log.info("after {} task {} finished, worker has been interrupted. not putting result in queue.", dispatcherName, finalCounter); @@ -108,12 +107,15 @@ public class Stage implements Ru 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; + boolean isRetryable = pipelineBackoff.isRetryable(e); + if (!isRetryable) { + log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e)); + pipeline.close(); + return null; + } + log.warn("{} RETRY AFTER ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e)); + pipelineBackoff.backoff(); } } log.error("tried for {} times, failed batch...", retryForSingleTask.get()); @@ -150,6 +152,10 @@ public class Stage implements Ru return false; } + public void startRunning() { + ioPool.submit(this); + } + public void close() { log.info("Stage {} is getting shutdown", dispatcher.getClass().getSimpleName()); ioPool.shutdownNow(); // прерываем обработчики @@ -160,16 +166,18 @@ public class Stage implements Ru if (consumer != null) { consumer.accept(b); } - q.put(b); + try (var m = stopWatch.start(dispatcherName + "-put")) { + q.put(b); + } } catch (InterruptedException e) { log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e)); Thread.currentThread().interrupt(); } } - X takeQuiet(BlockingQueue q, Duration timeout) { - try { - return q.poll(timeout.toMillis(), TimeUnit.MILLISECONDS); + X takeQuiet() { + try (var m = stopWatch.start(dispatcherName + "-take")) { + return in.take(); } catch (InterruptedException e) { log.error("Main thread for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e)); Thread.currentThread().interrupt(); @@ -185,4 +193,11 @@ public class Stage implements Ru this.consumer = consumer; } + public void setPipeline(Pipeline pipeline) { + this.pipeline = pipeline; + } + + public void setBackoff(PipelineBackoff pipelineBackoff) { + this.pipelineBackoff = pipelineBackoff; + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java index dca3bdf..450a87a 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java @@ -14,8 +14,8 @@ import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.builder.PersonFactoryBuilderV2; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.external_request.signals.Person; -import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.indic.PersonServiceV2; import ru.nbch.credit_tracker.service.task.Handler; @@ -25,12 +25,12 @@ import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2; @Slf4j public class FidIdEnricher implements Handler { private final PersonServiceV2 personService; - private final CreditTrackerSettings config; + private final CreditTrackerSettings creditTrackerSettings; public FidIdEnricher(PersonServiceV2 personService, - CreditTrackerSettings config) { + CreditTrackerSettings creditTrackerSettings) { this.personService = personService; - this.config = config; + this.creditTrackerSettings = creditTrackerSettings; } @Override @@ -49,7 +49,7 @@ public class FidIdEnricher implements Handler { } private Path writeSignals(List persons, Long packageId) { - Path dir = Paths.get(config.getLoadDir(), "tmp_folder"); + Path dir = Paths.get(creditTrackerSettings.getLoadDir(), "tmp_folder"); try { Files.createDirectories(dir); } catch (IOException e) { @@ -62,7 +62,7 @@ public class FidIdEnricher implements Handler { Path done = dir.resolve(base); try (OutputStream os = Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW)) { - SignalsStaxXmlWriterV2 writer = new SignalsStaxXmlWriterV2(persons, null, config.getEncoding()); + SignalsStaxXmlWriterV2 writer = new SignalsStaxXmlWriterV2(persons, null, creditTrackerSettings.getEncoding()); writer.writeWithStax(os); // перегрузи на OutputStream, если нужно } catch (IOException e) { throw new RuntimeException(e); @@ -75,6 +75,11 @@ public class FidIdEnricher implements Handler { return done; } + @Override + public int preferredConcurrency() { + return creditTrackerSettings.getPipeline().idEnricherStageConcurrency(); + } + @Override public PartFilePayload cookPoison() { return PartFilePayload.POISON; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java similarity index 93% rename from src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherV2.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java index 57318e1..a124648 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherV2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java @@ -20,16 +20,16 @@ import ru.nbch.credit_tracker.service.indic.FlagsService2; @Slf4j @Service -public class FlagsEnricherV2 implements Consumer { +public class FlagsEnricher implements Consumer { private final FlagsService2 flagsService; private final ExecutorService ioPool; - public FlagsEnricherV2(FlagsService2 flagsService, - PhoneFidMapMapper phoneFidMapMapper) { + public FlagsEnricher(FlagsService2 flagsService, + PhoneFidMapMapper phoneFidMapMapper) { this.flagsService = flagsService; - this.ioPool = new ThreadPoolExecutor(5, - 5, + this.ioPool = new ThreadPoolExecutor(10, + 10, 0L, TimeUnit.MINUTES, new LinkedBlockingQueue<>(100), r -> { 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 ee1c8e0..282bf47 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 @@ -33,7 +33,7 @@ public class KickPipeLineImpl implements IPipeLineStarter { this.packageId = packageId; this.pkgPath = pkgPath; this.xmlProcessor = xmlProcessor; - this.BATCH_SIZE = creditTrackerSettings.getBatchSize(); + this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize(); } private static class PipelineStarterException extends SignalClientException { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java index 034a11b..a8fc92e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java @@ -5,6 +5,7 @@ import java.util.List; import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.PhoneFidMap; import ru.nbch.credit_tracker.entities.indic.Name; import ru.nbch.credit_tracker.model.NameData; @@ -23,9 +24,12 @@ import ru.nbch.credit_tracker.utils.CTUtil; //todo опять нейминг класса, варианты: ResolveSubject; DefineIcrsSubject; IcrsMapping/Resolving; PackageSaver; Filler... public class SaveSubjectHandler implements Handler { private final PhoneService2 phoneService2; + private final CreditTrackerSettings creditTrackerSettings; - public SaveSubjectHandler(PhoneService2 phoneService2) { + public SaveSubjectHandler(PhoneService2 phoneService2, + CreditTrackerSettings creditTrackerSettings) { this.phoneService2 = phoneService2; + this.creditTrackerSettings = creditTrackerSettings; } @Override @@ -64,12 +68,12 @@ public class SaveSubjectHandler implements Handler pipeLineStarter; private IPipeLineTerminator pipeLineTerminator; private Semaphore semaphore; + private PipelineBackoff backoff; private PipeLineBuilderV2() { this.stages = new ArrayList<>(); @@ -60,12 +62,18 @@ public class PipeLineBuilderV2 backoff(PipelineBackoff backoff) { + this.backoff = backoff; + return this; + } + public Pipeline build() { Pipeline p = new Pipeline(); p.setSemaphore(semaphore); p.setStarter(() -> { stages.forEach(stage -> { stage.setPipeline(p); + stage.setBackoff(backoff); stage.startRunning(); }); pipeLineStarter.startFillingQueue(firstStage.getIn()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilderV2.java index 36df1aa..fefc256 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilderV2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilderV2.java @@ -46,6 +46,7 @@ public class StageBuilderV2 handlerResult(Consumer consumer) { this.consumer = Objects.requireNonNull(consumer, "consumer is required"); return this; diff --git a/src/main/java/ru/nbch/credit_tracker/utils/TaskTimer.java b/src/main/java/ru/nbch/credit_tracker/utils/TaskTimer.java new file mode 100644 index 0000000..7c4ec51 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/utils/TaskTimer.java @@ -0,0 +1,166 @@ +package ru.nbch.credit_tracker.utils; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.LongAdder; +import java.util.stream.Collectors; + +public class TaskTimer { + private final ConcurrentMap stats = new ConcurrentHashMap<>(); + + /** + * Начать измерение времени для задачи с именем taskName. + * Используется обычно через try-with-resources: + * + * try (var m = timer.start("dispatcher.handle")) { + * dispatcher.handle(b); + * } + */ + public Measurement start(String taskName) { + return new Measurement(this, taskName, System.nanoTime()); + } + + void record(String taskName, long durationNanos) { + TaskStats s = stats.computeIfAbsent(taskName, key -> new TaskStats()); + s.record(durationNanos); + } + + /** + * Снимок статистики по всем задачам на текущий момент. + */ + public Map snapshot() { + return stats.entrySet().stream() + .collect(Collectors.toUnmodifiableMap( + Map.Entry::getKey, + e -> e.getValue().snapshot() + )); + } + + /** + * Снимок статистики по одной задаче. + */ + public StatsSnapshot snapshot(String taskName) { + TaskStats s = stats.get(taskName); + return s == null ? StatsSnapshot.empty() : s.snapshot(); + } + + public String info() { + return snapshot().entrySet().stream().map(stringStatsSnapshotEntry -> + "%s: %s".formatted(stringStatsSnapshotEntry.getKey(), stringStatsSnapshotEntry.getValue())).collect(Collectors.joining("\n", "\n", "")); + } + + // ---------- Вспомогательные классы ---------- + + public static final class StatsSnapshot { + public final long count; + public final long totalNanos; + public final long minNanos; + public final long maxNanos; + + public StatsSnapshot(long count, long totalNanos, long minNanos, long maxNanos) { + this.count = count; + this.totalNanos = totalNanos; + this.minNanos = minNanos; + this.maxNanos = maxNanos; + } + + public static StatsSnapshot empty() { + return new StatsSnapshot(0, 0, 0, 0); + } + + public double avgMillis() { + return count == 0 ? 0.0 : (totalNanos / 1_000_000.0) / count; + } + + public double totalMillis() { + return totalNanos / 1_000_000.0; + } + + public double minMillis() { + return minNanos == 0 ? 0.0 : minNanos / 1_000_000.0; + } + + public double maxMillis() { + return maxNanos == 0 ? 0.0 : maxNanos / 1_000_000.0; + } + + @Override + public String toString() { + return "StatsSnapshot{" + + "count=" + count + + ", totalMillis=" + totalMillis() + + ", avgMillis=" + avgMillis() + + ", minMillis=" + minMillis() + + ", maxMillis=" + maxMillis() + + '}'; + } + } + + private static final class TaskStats { + private final LongAdder count = new LongAdder(); + private final LongAdder totalNanos = new LongAdder(); + private final AtomicLong minNanos = new AtomicLong(Long.MAX_VALUE); + private final AtomicLong maxNanos = new AtomicLong(Long.MIN_VALUE); + + void record(long durationNanos) { + count.increment(); + totalNanos.add(durationNanos); + + // min + for (;;) { + long current = minNanos.get(); + if (durationNanos >= current) { + break; + } + if (minNanos.compareAndSet(current, durationNanos)) { + break; + } + } + + // max + for (;;) { + long current = maxNanos.get(); + if (durationNanos <= current) { + break; + } + if (maxNanos.compareAndSet(current, durationNanos)) { + break; + } + } + } + + StatsSnapshot snapshot() { + long c = count.sum(); + if (c == 0) { + return StatsSnapshot.empty(); + } + long total = totalNanos.sum(); + long min = minNanos.get(); + long max = maxNanos.get(); + if (min == Long.MAX_VALUE) min = 0; + if (max == Long.MIN_VALUE) max = 0; + return new StatsSnapshot(c, total, min, max); + } + } + + public final class Measurement implements AutoCloseable { + private final String taskName; + private final long startNanos; + private boolean closed = false; + + private Measurement(TaskTimer timer, String taskName, long startNanos) { + this.taskName = taskName; + this.startNanos = startNanos; + } + + @Override + public void close() { + if (closed) return; + closed = true; + long duration = System.nanoTime() - startNanos; + TaskTimer.this.record(taskName, duration); + } + } +}