treshenkov/credit_tracker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add time statistic

Browse source
This commit is contained in:
etreschenkov 2025-11-21 21:00:12 +03:00
parent 7e641162ab
commit d06ea8334b
13 changed files with 302 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java
View file

@ -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;
}

18
src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java Normal file
View file

@ -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<Class<? extends Throwable>> retryableExceptions
) {
}

4
src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java
View file

@ -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<T extends StagePayload> implements FactoryBean<Bl
@Override
public BlockingQueue<T> getObject() throws Exception {
return new LinkedBlockingQueue<>(capacity);
return new ArrayBlockingQueue<>(capacity);
}
@Override

33
src/main/java/ru/nbch/credit_tracker/service/task/PipelineBackoff.java Normal file
View file

@ -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<Class<? extends Throwable>> 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()));
}
}

16
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java
View file

@ -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<SubjectPayload> saveSubjectQueueFactory;
private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
private final StageQueueFactory<PartFilePayload> 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<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> 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) {

61
src/main/java/ru/nbch/credit_tracker/service/task/Stage.java
View file

@ -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<X extends StagePayload, Z extends StagePayload> implements Ru
private Consumer<Z> 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<X extends StagePayload, Z extends StagePayload> implements Ru
public Stage(Handler<X, Z> dispatcher,
BlockingQueue<X> in,
BlockingQueue<Z> 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<X extends StagePayload, Z extends StagePayload> implements Ru
);
}
public void setPipeline(Pipeline pipeline) {
this.pipeline = pipeline;
}
public void startRunning() {
ioPool.submit(this);
}
@Override
public void run() {
List<Future<Z>> inflight = new ArrayList<>();
@ -69,11 +65,10 @@ public class Stage<X extends StagePayload, Z extends StagePayload> 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<X extends StagePayload, Z extends StagePayload> 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<X extends StagePayload, Z extends StagePayload> implements Ru
int finalCounter = counter;
AtomicInteger retryForSingleTask = new AtomicInteger(0);
Future<Z> 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<X extends StagePayload, Z extends StagePayload> 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<X extends StagePayload, Z extends StagePayload> 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<X extends StagePayload, Z extends StagePayload> 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<X> 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<X extends StagePayload, Z extends StagePayload> implements Ru
this.consumer = consumer;
}
public void setPipeline(Pipeline pipeline) {
this.pipeline = pipeline;
}
public void setBackoff(PipelineBackoff pipelineBackoff) {
this.pipelineBackoff = pipelineBackoff;
}
}

17
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java
View file

@ -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<SubjectPayload, PartFilePayload> {
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<SubjectPayload, PartFilePayload> {
}
private Path writeSignals(List<Person> 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<SubjectPayload, PartFilePayload> {
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<SubjectPayload, PartFilePayload> {
return done;
}
@Override
public int preferredConcurrency() {
return creditTrackerSettings.getPipeline().idEnricherStageConcurrency();
}
@Override
public PartFilePayload cookPoison() {
return PartFilePayload.POISON;

10
src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherV2.java → src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java
View file

@ -20,16 +20,16 @@ import ru.nbch.credit_tracker.service.indic.FlagsService2;
@Slf4j
@Service
public class FlagsEnricherV2 implements Consumer<SubjectPayload> {
public class FlagsEnricher implements Consumer<SubjectPayload> {
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 -> {

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java
View file

@ -33,7 +33,7 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
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 {

10
src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java
View file

@ -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<SubjectPayload, SubjectPayload> {
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<SubjectPayload, SubjectPayloa
for (int i = 0; i < subjects.size(); i++) {
subjects.get(i).setMapId(ids.get(i));
}
return new SubjectPayload(taskPayload.getPackageId(), subjects);
return taskPayload;
}
@Override
public int preferredConcurrency() {
return 15;
return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency();
}
@Override

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

@ -7,6 +7,7 @@ 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;
import ru.nbch.credit_tracker.service.task.Stage;
import ru.nbch.credit_tracker.service.task.setup.step.HandlerStep;
@ -17,6 +18,7 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
private IPipeLineStarter<FIRST> pipeLineStarter;
private IPipeLineTerminator<LAST, ?> pipeLineTerminator;
private Semaphore semaphore;
private PipelineBackoff backoff;
private PipeLineBuilderV2() {
this.stages = new ArrayList<>();
@ -60,12 +62,18 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
return this;
}
public PipeLineBuilderV2<FIRST, LAST> 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());

1
src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilderV2.java
View file

@ -46,6 +46,7 @@ public class StageBuilderV2<FIRST extends StagePayload, I extends StagePayload,
return this;
}
@Override
public BuildStep<FIRST, O> handlerResult(Consumer<O> consumer) {
this.consumer = Objects.requireNonNull(consumer, "consumer is required");
return this;

166
src/main/java/ru/nbch/credit_tracker/utils/TaskTimer.java Normal file
View file

@ -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<String, TaskStats> 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<String, StatsSnapshot> 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);
}
}
}