check-data-integrity v2

.gitignore

@@ -0,0 +1 @@
+/target/

pom.xml

@@ -0,0 +1,40 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>de.extio</groupId>
+    <artifactId>check-data-integrity2</artifactId>
+    <version>2.0.0</version>
+    <name>check-data-integrity</name>
+
+    <properties>
+        <java.version>21</java.version>
+        <maven.compiler.target>21</maven.compiler.target>
+        <maven.compiler.source>21</maven.compiler.source>        
+    </properties>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <shadedArtifactAttached>true</shadedArtifactAttached>
+                            <transformers>
+                                <transformer
+                                    implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+                                    <mainClass>checkDataIntegrity.Main</mainClass>
+                                </transformer>
+                            </transformers>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

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) {
+	
+}

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<Buffer> 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;
+		
+	}
+	
+}

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 <Path0> <Path1> <threads> <skip> <timeLimit ms>");
+			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;
+			}
+		});
+	}
+}

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) {
+	
+}

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<Path> {
+	
+	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<Buffer> queue0 = new LinkedBlockingQueue<>(1);
+				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 {
+					
+					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<Buffer> 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) {
+		}
+	}
+	
+}

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<Path> queue;
+	
+	public Worker(final Args args, final ProgramState state, final BlockingQueue<Path> 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) {
+		}
+	}
+	
+}

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<Worker> workers;
+	
+	private final BlockingQueue<Path> 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) {
+			}
+		}
+	}
+	
+}