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

pipeline bugfixes

Browse source
This commit is contained in:
ialbert 2025-11-11 16:59:18 +03:00
parent 395613ae01
commit 945694e50d
2 changed files with 42 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -4,7 +4,6 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
@ -22,8 +21,11 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
private final BlockingQueue<Z> out;
private final Z POISON;
private final ExecutorService ioPool;
private final CompletableFuture<Void> done = new CompletableFuture<>();
private Consumer<Z> consumer = null;
private final String dispatcherName;
private static final ThreadPoolExecutor.CallerRunsPolicy callerRunsPolicy
= new ThreadPoolExecutor.CallerRunsPolicy();
public Stage(Handler<X, Z> dispatcher,
BlockingQueue<X> in,
@ -32,24 +34,24 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
this.dispatcher = dispatcher;
this.in = in;
this.out = out;
this.dispatcherName = dispatcher.getClass().getSimpleName();
this.ioPool = new ThreadPoolExecutor(10,
10,
0L, TimeUnit.MINUTES,
new LinkedBlockingQueue<>(1),
r -> {
Thread t = new Thread(r);
t.setName("pipe-handler-%s-%d".formatted(dispatcher.getClass().getSimpleName(), t.threadId()));
t.setName("pipe-handler-%s-%d".formatted(dispatcherName, t.threadId()));
return t;
},
new ThreadPoolExecutor.CallerRunsPolicy()
(r, executor) -> {
log.info("stage {}: all threads are working; switch to CallerRunsPolicy", dispatcherName);
callerRunsPolicy.rejectedExecution(r, executor);
}
);
this.POISON = POISON;
}
public CompletableFuture<Void> doneFuture() {
return done;
}
public void startRunning() {
ioPool.submit(this);
}
@ -78,37 +80,48 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
}
putQuiet(out, POISON);
log.info("stage {} finished.", dispatcher.getClass().getSimpleName());
done.complete(null);
return;
}
counter++;
int finalCounter = counter;
Future<Z> f = ioPool.submit(() -> {
if (wasInterrupted()) {
log.info("stage {} task {} interrupted...", dispatcher.getClass().getSimpleName(), finalCounter);
try {
if (wasInterrupted()) {
log.info("stage {} task {} interrupted...", dispatcher.getClass().getSimpleName(), finalCounter);
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.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter);
Z p = dispatcher.handle(b);
putQuiet(out, p);
log.info("stage {} task {} finished.", dispatcher.getClass().getSimpleName(), finalCounter);
return p;
}
);
inflight.add(f);
}
} catch (Exception e) {
log.error("error: {}", ExceptionUtils.getStackTrace(e));
log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
putQuiet(out, POISON);
done.complete(null);
}
}
private String info() {
ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool;
return "executor active tasks: %d; queue size: %d, completed tasks: %d".formatted(
casted.getActiveCount(),
casted.getQueue().size(),
casted.getCompletedTaskCount()
);
}
private boolean wasInterrupted() {
if (Thread.currentThread().isInterrupted()) {
log.warn("pipeline was interrupted...");
putQuiet(out, POISON);
done.complete(null);
return true;
}
return false;
@ -117,9 +130,6 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
public void close() {
log.info("Stage {} is getting shutdown", dispatcher.getClass().getSimpleName());
ioPool.shutdownNow(); // прерываем обработчики
if (!done.isDone()) {
done.completeExceptionally(new java.util.concurrent.CancellationException("Stage cancelled"));
}
}
void putQuiet(BlockingQueue<Z> q, Z b) {
@ -129,7 +139,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
}
q.put(b);
} catch (InterruptedException e) {
log.error("Error: {}", ExceptionUtils.getStackTrace(e));
log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
Thread.currentThread().interrupt();
}
}
@ -138,7 +148,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
try {
return q.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.error("Error: {}", ExceptionUtils.getStackTrace(e));
log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
Thread.currentThread().interrupt();
return null;
}

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

@ -15,6 +15,7 @@ import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
private final Logger log = LoggerFactory.getLogger(getClass());
@ -51,6 +52,7 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
try {
firstQueue.put(new SubjectPayload(packageId, subjectProfileBatch));
} catch (InterruptedException e) {
log.error("{}", ExceptionUtils.getStackTrace(e));
throw new RuntimeException(e);
}
request.getSubjects().getSubject().clear();
@ -60,10 +62,14 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
,
BATCH_SIZE, "/MonitoringRequest/Subjects/Subject");
} catch (IOException e) {
throw new RuntimeException("Error while parsing monitoring request", e);
log.error("{}", ExceptionUtils.getStackTrace(e));
} finally {
log.debug("Send poison pill");
firstQueue.add(new SubjectPayload(true));
try {
firstQueue.put(new SubjectPayload(true));
} catch (InterruptedException e) {
log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e));
}
}
}
}