checkDataIntegrity - major improved version;

git-svn-id: svn://home0/checkDataIntegrity@3514 71dcc7fd-b1f2-4e5d-bb8e-274146e81173

src/checkDataIntegrity/Args.java

@@ -0,0 +1,6 @@
+package checkDataIntegrity;
+
+
+record Args(String path0, String path1, int skip, int timeLimit) {
+	
+}

src/checkDataIntegrity/BufferPool.java

@@ -0,0 +1,21 @@
+package checkDataIntegrity;
+
+import java.util.concurrent.LinkedBlockingQueue;
+
+class BufferPool {
+	
+	private static final LinkedBlockingQueue<byte[]> 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);
+	}
+	
+}

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 {
+	public static void main(final String[] args) throws Exception {
-		if (args.length < 4) {
+		final Args parsedArgs = parseArgs(args);
-			System.err.println("Usage: checkDataIntegrity.java <Path0> <Path1> <skip> <limit>");
+		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 ProgramState state = new ProgramState(System.currentTimeMillis(), new AtomicLong(), new AtomicLong(), new AtomicLong());
-		final AtomicLong count = new AtomicLong();
 		
-		System.out.println("Scanning Tree");
+		final WorkerPool workerPool = new WorkerPool(parsedArgs, state, 32);
 		
-		final List<Path> paths = new ArrayList<>(limit);
+		final AtomicLong walkTreeCounter = new AtomicLong();
-		Files.walkFileTree(Path.of(path0), new FileVisitor<>() {
+		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) {
+				if (attrs.isRegularFile() && walkTreeCounter.incrementAndGet() > parsedArgs.skip()) {
-					paths.add(file);
+					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);
+		workerPool.awaitEmpty();
-		final AtomicLong countErr = new AtomicLong();
+		workerPool.interrupt();
-		final ThreadLocal<byte[]> bufs0 = ThreadLocal.withInitial(() -> new byte[8192]);
+		workerPool.join();
-		final ThreadLocal<byte[]> bufs1 = ThreadLocal.withInitial(() -> new byte[8192]);
 		
-		try {
+		printStats(state);
-			paths
+		
-					.stream()
+		System.exit(state.countErr().get() > 0 ? 1 : 0);
-					.parallel()
+	}
-					.forEach(p0 -> {
+	
-						try {
+	private static Args parseArgs(final String[] args) {
-							final long cnt = count.incrementAndGet();
+		if (args.length < 4) {
-							if (cnt % 1000 == 0) {
+			System.err.println("Usage: checkDataIntegrity.java <Path0> <Path1> <skip> <timeLimit ms>");
-								System.out.println(cnt);
+			return null;
-							}
-							
-							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();
 		}
 		
-		System.out.println("Took " + Duration.ofMillis(System.currentTimeMillis() - start).toString());
+		final Args result = new Args(
-		System.out.println(((count.get() > 0) ? count.get() : "No") + " files checked");
+				args[0],
-		System.out.println(countErr.get() + " errors");
+				args[1],
-		
+				Integer.parseInt(args[2]),
-		System.exit(countErr.get() > 0 ? 1 : 0);
+				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());
 	}
 	
 }

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

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

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