Fix: Ensure blocking queues are local & cancel futures on early exit

The original code used class-level `BlockingQueue` instances (`queue0`, `queue1`) which caused issues when processing multiple files concurrently. Each comparison needed its *own* dedicated queues to avoid data corruption and incorrect comparisons between different file pairs. This commit creates the `BlockingQueue` instances locally within the file comparison loop.

Additionally, the futures submitted to the executor service were not always being cancelled if an exception occurred or the comparison finished before reaching the end of the files.  This could lead to lingering threads. The `finally` block now ensures that both futures are cancelled regardless of how the try block exits, preventing resource leaks and potential deadlocks.


git-svn-id: svn://home0/checkDataIntegrity@3745 71dcc7fd-b1f2-4e5d-bb8e-274146e81173
This commit is contained in:
(no author)
2025-06-20 11:49:38 +00:00
parent 2a1ff2b30a
commit d3134c05a5
3 changed files with 9 additions and 11 deletions

Binary file not shown.

Binary file not shown.

View File

@@ -24,10 +24,6 @@ class Task implements Consumer<Path> {
private final ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor(); private final ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();
private final BlockingQueue<Buffer> queue0 = new LinkedBlockingQueue<>(1);
private final BlockingQueue<Buffer> queue1 = new LinkedBlockingQueue<>(1);
Task(final Args args, final ProgramState state) { Task(final Args args, final ProgramState state) {
this.args = args; this.args = args;
this.state = state; this.state = state;
@@ -45,12 +41,15 @@ class Task implements Consumer<Path> {
final Path p1 = Path.of(this.args.path1() + rel); final Path p1 = Path.of(this.args.path1() + rel);
try (RandomAccessFile raf0 = new RandomAccessFile(path.toFile(), "r"); RandomAccessFile raf1 = new RandomAccessFile(p1.toFile(), "r")) { try (RandomAccessFile raf0 = new RandomAccessFile(path.toFile(), "r"); RandomAccessFile raf1 = new RandomAccessFile(p1.toFile(), "r")) {
final Future<?> future0 = this.executorService.submit(() -> this.loader(raf0, this.queue0)); final BlockingQueue<Buffer> queue0 = new LinkedBlockingQueue<>(1);
final Future<?> future1 = this.executorService.submit(() -> this.loader(raf1, this.queue1)); final BlockingQueue<Buffer> queue1 = new LinkedBlockingQueue<>(1);
final Future<?> future0 = this.executorService.submit(() -> this.loader(raf0, queue0));
final Future<?> future1 = this.executorService.submit(() -> this.loader(raf1, queue1));
try { try {
while (true) { while (true) {
final Buffer buf0 = this.queue0.take(); final Buffer buf0 = queue0.take();
final Buffer buf1 = this.queue1.take(); final Buffer buf1 = queue1.take();
try { try {
if (buf0.len != buf1.len) { if (buf0.len != buf1.len) {
System.out.println(rel + ": Length mismatch"); System.out.println(rel + ": Length mismatch");
@@ -73,16 +72,15 @@ class Task implements Consumer<Path> {
} }
} }
} }
catch (final Exception ex) { finally {
future0.cancel(true); future0.cancel(true);
future1.cancel(true); future1.cancel(true);
throw ex;
} }
} }
} }
catch (NoSuchFileException | FileNotFoundException | InvalidPathException ex) { catch (NoSuchFileException | FileNotFoundException | InvalidPathException ex) {
} }
catch (InterruptedException ex) { catch (final InterruptedException ex) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
} }
catch (final Exception ex) { catch (final Exception ex) {