diff --git a/bin/checkDataIntegrity/BufferPool$Buffer.class b/bin/checkDataIntegrity/BufferPool$Buffer.class new file mode 100644 index 0000000..8a3bfa3 Binary files /dev/null and b/bin/checkDataIntegrity/BufferPool$Buffer.class differ diff --git a/bin/checkDataIntegrity/BufferPool.class b/bin/checkDataIntegrity/BufferPool.class index 2b15b54..d3bb0e4 100644 Binary files a/bin/checkDataIntegrity/BufferPool.class and b/bin/checkDataIntegrity/BufferPool.class differ diff --git a/bin/checkDataIntegrity/Main.class b/bin/checkDataIntegrity/Main.class index edd00b2..5355584 100644 Binary files a/bin/checkDataIntegrity/Main.class and b/bin/checkDataIntegrity/Main.class differ diff --git a/bin/checkDataIntegrity/Task.class b/bin/checkDataIntegrity/Task.class index 931e1a4..a8cc54b 100644 Binary files a/bin/checkDataIntegrity/Task.class and b/bin/checkDataIntegrity/Task.class differ diff --git a/checkDataIntegrity.jar b/checkDataIntegrity.jar index a425292..626e404 100644 Binary files a/checkDataIntegrity.jar and b/checkDataIntegrity.jar differ diff --git a/src/checkDataIntegrity/BufferPool.java b/src/checkDataIntegrity/BufferPool.java index 5b5a3ae..3372071 100644 --- a/src/checkDataIntegrity/BufferPool.java +++ b/src/checkDataIntegrity/BufferPool.java @@ -4,18 +4,26 @@ import java.util.concurrent.LinkedBlockingQueue; class BufferPool { - private static final LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); + private static final LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); - static byte[] get() { - byte[] buf = queue.poll(); + static Buffer get() { + Buffer buf = queue.poll(); if (buf == null) { - buf = new byte[4096]; + buf = new Buffer(); } return buf; } - static void put(final byte[] buf) { + static void put(final Buffer buf) { queue.offer(buf); } + static class Buffer { + + byte[] buf = new byte[16384]; + + volatile int len; + + } + } diff --git a/src/checkDataIntegrity/Main.java b/src/checkDataIntegrity/Main.java index f1e31c3..dc587bc 100644 --- a/src/checkDataIntegrity/Main.java +++ b/src/checkDataIntegrity/Main.java @@ -32,7 +32,7 @@ public class Main { } private static Args parseArgs(final String[] args) { - if (args.length < 4) { + if (args.length < 5) { System.err.println("Usage: checkDataIntegrity.java "); return null; } diff --git a/src/checkDataIntegrity/Task.java b/src/checkDataIntegrity/Task.java index fe929b3..1a0105e 100644 --- a/src/checkDataIntegrity/Task.java +++ b/src/checkDataIntegrity/Task.java @@ -1,19 +1,33 @@ package checkDataIntegrity; import java.io.FileNotFoundException; +import java.io.IOException; import java.io.RandomAccessFile; import java.nio.file.InvalidPathException; import java.nio.file.NoSuchFileException; import java.nio.file.Path; import java.util.Arrays; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; import java.util.function.Consumer; +import checkDataIntegrity.BufferPool.Buffer; + class Task implements Consumer { private final Args args; private final ProgramState state; + private final ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor(); + + private final BlockingQueue queue0 = new LinkedBlockingQueue<>(1); + + private final BlockingQueue queue1 = new LinkedBlockingQueue<>(1); + Task(final Args args, final ProgramState state) { this.args = args; this.state = state; @@ -30,42 +44,77 @@ class Task implements Consumer { final String rel = path.toString().substring(this.args.path0().length()); final Path p1 = Path.of(this.args.path1() + rel); - int len0, len1; - final byte[] buf0 = BufferPool.get(); - final byte[] buf1 = BufferPool.get(); - try ( - RandomAccessFile raf0 = new RandomAccessFile(path.toFile(), "r"); - RandomAccessFile raf1 = new RandomAccessFile(p1.toFile(), "r")) { - - do { - len0 = raf0.read(buf0); - len1 = raf1.read(buf1); - if (len0 != len1) { - System.out.println(rel + ": Length mismatch"); - this.state.countErr().incrementAndGet(); - break; - } - if (len0 > -1) { - this.state.bytes().addAndGet(len0); - if (!Arrays.equals(buf0, 0, len0, buf1, 0, len1)) { - System.out.println(rel + ": Data mismatch"); - this.state.countErr().incrementAndGet(); - break; + 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 Future future1 = this.executorService.submit(() -> this.loader(raf1, this.queue1)); + try { + while (true) { + final Buffer buf0 = this.queue0.take(); + final Buffer buf1 = this.queue1.take(); + try { + if (buf0.len != buf1.len) { + System.out.println(rel + ": Length mismatch"); + this.state.countErr().incrementAndGet(); + break; + } + if (buf0.len == -1) { + break; + } + this.state.bytes().addAndGet(buf0.len); + if (!Arrays.equals(buf0.buf, 0, buf0.len, buf1.buf, 0, buf1.len)) { + System.out.println(rel + ": Data mismatch"); + this.state.countErr().incrementAndGet(); + break; + } + } + finally { + BufferPool.put(buf0); + BufferPool.put(buf1); } } - } while (len0 > -1); - } - finally { - BufferPool.put(buf0); - BufferPool.put(buf1); + } + catch (final Exception ex) { + future0.cancel(true); + future1.cancel(true); + throw ex; + } } } catch (NoSuchFileException | FileNotFoundException | InvalidPathException ex) { } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } catch (final Exception ex) { System.out.println(path + ": Exception: " + ex.getClass().getName() + " " + ex.getMessage()); this.state.countErr().incrementAndGet(); } } + private void loader(final RandomAccessFile raf, final BlockingQueue queue) { + try { + while (!Thread.currentThread().isInterrupted()) { + final Buffer buf = BufferPool.get(); + buf.len = raf.read(buf.buf); + if (buf.len == -1) { + queue.put(buf); + return; + } + queue.put(buf); + } + } + catch (final IOException e) { + e.printStackTrace(); + final Buffer buf = BufferPool.get(); + buf.len = -1; + try { + queue.put(buf); + } + catch (final InterruptedException e1) { + } + } + catch (final InterruptedException e) { + } + } + }