diff --git a/bin/checkDataIntegrity/Args.class b/bin/checkDataIntegrity/Args.class new file mode 100644 index 0000000..5a2fc18 Binary files /dev/null and b/bin/checkDataIntegrity/Args.class differ diff --git a/bin/checkDataIntegrity/BufferPool.class b/bin/checkDataIntegrity/BufferPool.class new file mode 100644 index 0000000..444dc5c Binary files /dev/null and b/bin/checkDataIntegrity/BufferPool.class differ diff --git a/bin/checkDataIntegrity/Main$1.class b/bin/checkDataIntegrity/Main$1.class index 473a9ba..fa4361b 100644 Binary files a/bin/checkDataIntegrity/Main$1.class and b/bin/checkDataIntegrity/Main$1.class differ diff --git a/bin/checkDataIntegrity/Main.class b/bin/checkDataIntegrity/Main.class index b333cba..99724f1 100644 Binary files a/bin/checkDataIntegrity/Main.class and b/bin/checkDataIntegrity/Main.class differ diff --git a/bin/checkDataIntegrity/ProgramState.class b/bin/checkDataIntegrity/ProgramState.class new file mode 100644 index 0000000..304fda4 Binary files /dev/null and b/bin/checkDataIntegrity/ProgramState.class differ diff --git a/bin/checkDataIntegrity/Task.class b/bin/checkDataIntegrity/Task.class new file mode 100644 index 0000000..931e1a4 Binary files /dev/null and b/bin/checkDataIntegrity/Task.class differ diff --git a/bin/checkDataIntegrity/Worker.class b/bin/checkDataIntegrity/Worker.class new file mode 100644 index 0000000..dae0973 Binary files /dev/null and b/bin/checkDataIntegrity/Worker.class differ diff --git a/bin/checkDataIntegrity/WorkerPool.class b/bin/checkDataIntegrity/WorkerPool.class new file mode 100644 index 0000000..ea625e8 Binary files /dev/null and b/bin/checkDataIntegrity/WorkerPool.class differ diff --git a/bin/module-info.class b/bin/module-info.class index 52a3e4c..d68e9cd 100644 Binary files a/bin/module-info.class and b/bin/module-info.class differ diff --git a/checkDataIntegrity.jar b/checkDataIntegrity.jar index 7a1a7f5..36619df 100644 Binary files a/checkDataIntegrity.jar and b/checkDataIntegrity.jar differ diff --git a/src/checkDataIntegrity/Args.java b/src/checkDataIntegrity/Args.java new file mode 100644 index 0000000..2ca1246 --- /dev/null +++ b/src/checkDataIntegrity/Args.java @@ -0,0 +1,6 @@ +package checkDataIntegrity; + + +record Args(String path0, String path1, int skip, int timeLimit) { + +} diff --git a/src/checkDataIntegrity/BufferPool.java b/src/checkDataIntegrity/BufferPool.java new file mode 100644 index 0000000..c9bf12e --- /dev/null +++ b/src/checkDataIntegrity/BufferPool.java @@ -0,0 +1,21 @@ +package checkDataIntegrity; + +import java.util.concurrent.LinkedBlockingQueue; + +class BufferPool { + + private static final LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); + + static byte[] get() { + byte[] buf = queue.poll(); + if (buf == null) { + buf = new byte[16384]; + } + return buf; + } + + static void put(final byte[] buf) { + queue.offer(buf); + } + +} diff --git a/src/checkDataIntegrity/Main.java b/src/checkDataIntegrity/Main.java index 1822296..2aec56a 100644 --- a/src/checkDataIntegrity/Main.java +++ b/src/checkDataIntegrity/Main.java @@ -1,47 +1,28 @@ package checkDataIntegrity; -import java.io.BufferedInputStream; -import java.io.FileInputStream; -import java.io.FileNotFoundException; import java.io.IOException; -import java.io.InputStream; import java.nio.file.FileVisitResult; import java.nio.file.FileVisitor; import java.nio.file.Files; -import java.nio.file.InvalidPathException; -import java.nio.file.NoSuchFileException; import java.nio.file.Path; import java.nio.file.attribute.BasicFileAttributes; import java.time.Duration; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; import java.util.concurrent.atomic.AtomicLong; public class Main { - public static void main(final String[] args) throws IOException { - if (args.length < 4) { - System.err.println("Usage: checkDataIntegrity.java "); + public static void main(final String[] args) throws Exception { + final Args parsedArgs = parseArgs(args); + if (parsedArgs == null) { return; } - final String path0 = args[0]; - final String path1 = args[1]; - final int skip = Integer.parseInt(args[2]); - final int limit = Integer.parseInt(args[3]); - final int max = skip + limit; - System.out.println("Path0: " + path0); - System.out.println("Path1: " + path1); - System.out.println("Skip: " + skip); - System.out.println("Limit: " + limit); - final long start = System.currentTimeMillis(); - final AtomicLong count = new AtomicLong(); + final ProgramState state = new ProgramState(System.currentTimeMillis(), new AtomicLong(), new AtomicLong(), new AtomicLong()); - System.out.println("Scanning Tree"); + final WorkerPool workerPool = new WorkerPool(parsedArgs, state, 32); - final List paths = new ArrayList<>(limit); - Files.walkFileTree(Path.of(path0), new FileVisitor<>() { + final AtomicLong walkTreeCounter = new AtomicLong(); + Files.walkFileTree(Path.of(parsedArgs.path0()), new FileVisitor<>() { @Override public FileVisitResult preVisitDirectory(final Path dir, final BasicFileAttributes attrs) throws IOException { @@ -50,85 +31,65 @@ public class Main { @Override public FileVisitResult visitFile(final Path file, final BasicFileAttributes attrs) throws IOException { - if (attrs.isRegularFile() && count.incrementAndGet() > skip) { - paths.add(file); + if (attrs.isRegularFile() && walkTreeCounter.incrementAndGet() > parsedArgs.skip()) { + try { + workerPool.put(file); + } + catch (final InterruptedException e) { + return FileVisitResult.TERMINATE; + } } - return count.get() < max ? FileVisitResult.CONTINUE : FileVisitResult.TERMINATE; + + final boolean timeLimitReached = System.currentTimeMillis() - state.start() >= parsedArgs.timeLimit(); + if (timeLimitReached) { + System.out.println("Time limit reached"); + } + return timeLimitReached ? FileVisitResult.TERMINATE : FileVisitResult.CONTINUE; } @Override public FileVisitResult visitFileFailed(final Path file, final IOException exc) throws IOException { - // System.out.println(exc.getClass().getSimpleName() + " " + exc.getMessage()); return FileVisitResult.CONTINUE; } @Override public FileVisitResult postVisitDirectory(final Path dir, final IOException exc) throws IOException { - // if (exc != null) { - // System.out.println(exc.getClass().getSimpleName() + " " + exc.getMessage()); - // } return FileVisitResult.CONTINUE; } }); - count.set(0); - final AtomicLong countErr = new AtomicLong(); - final ThreadLocal bufs0 = ThreadLocal.withInitial(() -> new byte[8192]); - final ThreadLocal bufs1 = ThreadLocal.withInitial(() -> new byte[8192]); + workerPool.awaitEmpty(); + workerPool.interrupt(); + workerPool.join(); - try { - paths - .stream() - .parallel() - .forEach(p0 -> { - try { - final long cnt = count.incrementAndGet(); - if (cnt % 1000 == 0) { - System.out.println(cnt); - } - - final String rel = p0.toString().substring(path0.length()); - final Path p1 = Path.of(path1 + rel); - - final byte[] buf0 = bufs0.get(); - final byte[] buf1 = bufs1.get(); - try (InputStream inputStream0 = new BufferedInputStream(new FileInputStream(p0.toFile()), 65536); InputStream inputStream1 = new BufferedInputStream(new FileInputStream(p1.toFile()), 65536)) { - int len0, len1; - do { - len0 = inputStream0.read(buf0); - len1 = inputStream1.read(buf1); - if (len0 != len1) { - System.out.println(rel + ": Length mismatch"); - countErr.incrementAndGet(); - break; - } - if (len0 > -1 && !Arrays.equals(buf0, 0, len0, buf1, 0, len1)) { - System.out.println(rel + ": Data mismatch"); - countErr.incrementAndGet(); - break; - } - } while (len0 > -1); - } - } - catch (NoSuchFileException | FileNotFoundException | InvalidPathException ex) { - // System.out.println(rel + ": NoSuchFileException"); - } - catch (final Exception ex) { - System.out.println(p0 + ": Exception: " + ex.getClass().getName() + " " + ex.getMessage()); - countErr.incrementAndGet(); - } - }); - } - catch (final Exception ex) { - System.out.println("General Exception: " + ex.getClass().getName() + " " + ex.getMessage()); - countErr.incrementAndGet(); + printStats(state); + + System.exit(state.countErr().get() > 0 ? 1 : 0); + } + + private static Args parseArgs(final String[] args) { + if (args.length < 4) { + System.err.println("Usage: checkDataIntegrity.java "); + return null; } - System.out.println("Took " + Duration.ofMillis(System.currentTimeMillis() - start).toString()); - System.out.println(((count.get() > 0) ? count.get() : "No") + " files checked"); - System.out.println(countErr.get() + " errors"); - - System.exit(countErr.get() > 0 ? 1 : 0); + final Args result = new Args( + args[0], + args[1], + Integer.parseInt(args[2]), + Integer.parseInt(args[3])); + System.out.println(result); + return result; + } + + private static void printStats(final ProgramState state) { + final Duration duration = Duration.ofMillis(System.currentTimeMillis() - state.start()); + System.out.println("Took: " + duration.toString()); + System.out.println("Files checked: " + state.count().get()); + if (duration.toSeconds() > 0) { + System.out.println("Throughput: " + (state.bytes().get() / duration.toSeconds() / 1024 / 1024) + " Mib/s"); + } + System.out.println("Errors: " + state.countErr().get()); } } diff --git a/src/checkDataIntegrity/ProgramState.java b/src/checkDataIntegrity/ProgramState.java new file mode 100644 index 0000000..feff392 --- /dev/null +++ b/src/checkDataIntegrity/ProgramState.java @@ -0,0 +1,7 @@ +package checkDataIntegrity; + +import java.util.concurrent.atomic.AtomicLong; + +record ProgramState(long start, AtomicLong count, AtomicLong countErr, AtomicLong bytes) { + +} diff --git a/src/checkDataIntegrity/Task.java b/src/checkDataIntegrity/Task.java new file mode 100644 index 0000000..fe929b3 --- /dev/null +++ b/src/checkDataIntegrity/Task.java @@ -0,0 +1,71 @@ +package checkDataIntegrity; + +import java.io.FileNotFoundException; +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.function.Consumer; + +class Task implements Consumer { + + private final Args args; + + private final ProgramState state; + + Task(final Args args, final ProgramState state) { + this.args = args; + this.state = state; + } + + @Override + public void accept(final Path path) { + try { + final long cnt = this.state.count().incrementAndGet(); + if (cnt % 1000 == 0) { + System.out.println(cnt); + } + + 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; + } + } + } while (len0 > -1); + } + finally { + BufferPool.put(buf0); + BufferPool.put(buf1); + } + } + catch (NoSuchFileException | FileNotFoundException | InvalidPathException ex) { + } + catch (final Exception ex) { + System.out.println(path + ": Exception: " + ex.getClass().getName() + " " + ex.getMessage()); + this.state.countErr().incrementAndGet(); + } + } + +} diff --git a/src/checkDataIntegrity/Worker.java b/src/checkDataIntegrity/Worker.java new file mode 100644 index 0000000..c7f885d --- /dev/null +++ b/src/checkDataIntegrity/Worker.java @@ -0,0 +1,30 @@ +package checkDataIntegrity; + +import java.nio.file.Path; +import java.util.concurrent.BlockingQueue; + +class Worker extends Thread { + + private final Task task; + + private final BlockingQueue queue; + + public Worker(final Args args, final ProgramState state, final BlockingQueue queue) { + super(); + this.queue = queue; + this.task = new Task(args, state); + } + + @Override + public void run() { + try { + while (!Thread.interrupted()) { + final Path p = this.queue.take(); + this.task.accept(p); + } + } + catch (final InterruptedException e) { + } + } + +} diff --git a/src/checkDataIntegrity/WorkerPool.java b/src/checkDataIntegrity/WorkerPool.java new file mode 100644 index 0000000..036291e --- /dev/null +++ b/src/checkDataIntegrity/WorkerPool.java @@ -0,0 +1,62 @@ +package checkDataIntegrity; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +class WorkerPool { + + private final List workers; + + private final BlockingQueue queue; + + WorkerPool(final Args args, final ProgramState state, final int threads) { + System.out.println("Threads: " + threads); + + this.queue = new LinkedBlockingQueue<>(threads * 2); + + this.workers = new ArrayList<>(threads); + for (int i = 0; i < threads; i++) { + final Worker worker = new Worker(args, state, this.queue); + worker.setDaemon(true); + worker.setName("Worker #" + i); + worker.start(); + + this.workers.add(worker); + } + } + + void put(final Path path) throws InterruptedException { + this.queue.put(path); + } + + void awaitEmpty() { + while (!this.queue.isEmpty()) { + try { + Thread.sleep(10); + } + catch (final InterruptedException e) { + break; + } + } + } + + void interrupt() { + for (final Worker worker : this.workers) { + worker.interrupt(); + } + } + + void join() { + for (final Worker worker : this.workers) { + try { + worker.join(); + } + catch (final InterruptedException e) { + } + } + } + +}