diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b83d222 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/target/ diff --git a/pom.xml b/pom.xml new file mode 100644 index 0000000..c49d13c --- /dev/null +++ b/pom.xml @@ -0,0 +1,40 @@ + + 4.0.0 + de.extio + check-data-integrity2 + 2.0.0 + check-data-integrity + + + 21 + 21 + 21 + + + + + + org.apache.maven.plugins + maven-shade-plugin + + + + shade + + + true + + + checkDataIntegrity.Main + + + + + + + + + \ No newline at end of file diff --git a/src/main/java/de/extio/checkdataintegrity/Args.java b/src/main/java/de/extio/checkdataintegrity/Args.java new file mode 100644 index 0000000..bb8196c --- /dev/null +++ b/src/main/java/de/extio/checkdataintegrity/Args.java @@ -0,0 +1,5 @@ +package de.extio.checkdataintegrity; + +record Args(String path0, String path1, int threads, int skip, int timeLimit) { + +} diff --git a/src/main/java/de/extio/checkdataintegrity/BufferPool.java b/src/main/java/de/extio/checkdataintegrity/BufferPool.java new file mode 100644 index 0000000..5105feb --- /dev/null +++ b/src/main/java/de/extio/checkdataintegrity/BufferPool.java @@ -0,0 +1,29 @@ +package de.extio.checkdataintegrity; + +import java.util.concurrent.LinkedBlockingQueue; + +class BufferPool { + + private static final LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); + + static Buffer get() { + Buffer buf = queue.poll(); + if (buf == null) { + buf = new Buffer(); + } + return 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/main/java/de/extio/checkdataintegrity/Main.java b/src/main/java/de/extio/checkdataintegrity/Main.java new file mode 100644 index 0000000..da092c0 --- /dev/null +++ b/src/main/java/de/extio/checkdataintegrity/Main.java @@ -0,0 +1,98 @@ +package de.extio.checkdataintegrity; + +import java.io.IOException; +import java.nio.file.FileVisitResult; +import java.nio.file.FileVisitor; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.BasicFileAttributes; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicLong; + +public class Main { + + public static void main(final String[] args) throws Exception { + final Args parsedArgs = parseArgs(args); + if (parsedArgs == null) { + return; + } + + final ProgramState state = new ProgramState(System.currentTimeMillis(), new AtomicLong(), new AtomicLong(), new AtomicLong()); + final WorkerPool workerPool = new WorkerPool(parsedArgs, state, parsedArgs.threads()); + + walk(parsedArgs, state, workerPool); + + workerPool.awaitEmpty(); + workerPool.interrupt(); + workerPool.join(); + + printStats(state); + + System.exit(state.countErr().get() > 0 ? 1 : 0); + } + + private static Args parseArgs(final String[] args) { + if (args.length < 5) { + System.err.println("Usage: checkDataIntegrity.java "); + return null; + } + + final Args result = new Args( + args[0], + args[1], + Integer.parseInt(args[2]), + Integer.parseInt(args[3]), + Integer.parseInt(args[4])); + 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()); + } + + private static void walk(final Args parsedArgs, final ProgramState state, final WorkerPool workerPool) throws IOException { + 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 { + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult visitFile(final Path file, final BasicFileAttributes attrs) throws IOException { + if (attrs.isRegularFile() && walkTreeCounter.incrementAndGet() > parsedArgs.skip()) { + try { + workerPool.put(file); + } + catch (final InterruptedException e) { + return 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 { + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult postVisitDirectory(final Path dir, final IOException exc) throws IOException { + return FileVisitResult.CONTINUE; + } + }); + } +} diff --git a/src/main/java/de/extio/checkdataintegrity/ProgramState.java b/src/main/java/de/extio/checkdataintegrity/ProgramState.java new file mode 100644 index 0000000..0c03e15 --- /dev/null +++ b/src/main/java/de/extio/checkdataintegrity/ProgramState.java @@ -0,0 +1,7 @@ +package de.extio.checkdataintegrity; + +import java.util.concurrent.atomic.AtomicLong; + +record ProgramState(long start, AtomicLong count, AtomicLong countErr, AtomicLong bytes) { + +} diff --git a/src/main/java/de/extio/checkdataintegrity/Task.java b/src/main/java/de/extio/checkdataintegrity/Task.java new file mode 100644 index 0000000..38be008 --- /dev/null +++ b/src/main/java/de/extio/checkdataintegrity/Task.java @@ -0,0 +1,118 @@ +package de.extio.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 de.extio.checkdataintegrity.BufferPool.Buffer; + +class Task implements Consumer { + + private final Args args; + + private final ProgramState state; + + private final ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor(); + + 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); + + try (RandomAccessFile raf0 = new RandomAccessFile(path.toFile(), "r"); RandomAccessFile raf1 = new RandomAccessFile(p1.toFile(), "r")) { + final BlockingQueue queue0 = new LinkedBlockingQueue<>(1); + final BlockingQueue 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 { + + while (true) { + final Buffer buf0 = queue0.take(); + final Buffer buf1 = 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); + } + } + } + finally { + future0.cancel(true); + future1.cancel(true); + } + } + } + catch (NoSuchFileException | FileNotFoundException | InvalidPathException ex) { + } + catch (final 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) { + } + } + +} diff --git a/src/main/java/de/extio/checkdataintegrity/Worker.java b/src/main/java/de/extio/checkdataintegrity/Worker.java new file mode 100644 index 0000000..bfd6981 --- /dev/null +++ b/src/main/java/de/extio/checkdataintegrity/Worker.java @@ -0,0 +1,30 @@ +package de.extio.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/main/java/de/extio/checkdataintegrity/WorkerPool.java b/src/main/java/de/extio/checkdataintegrity/WorkerPool.java new file mode 100644 index 0000000..13e2d39 --- /dev/null +++ b/src/main/java/de/extio/checkdataintegrity/WorkerPool.java @@ -0,0 +1,62 @@ +package de.extio.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) { + } + } + } + +}