Introduce virtual threads and blocking queues for concurrent file comparison

This commit introduces a significant refactoring of the `Task` class to improve performance and resource utilization when comparing files.  Instead of reading directly within the main thread, it now leverages Java's virtual threads and blocking queues to handle file loading concurrently.

Key changes:

- **Virtual Threads:** Uses `ExecutorService` with `newVirtualThreadPerTaskExecutor()` to create lightweight virtual threads for reading from each file. This allows for highly concurrent I/O operations without excessive overhead.
- **Blocking Queues:** Employs `LinkedBlockingQueue` to facilitate communication between the file loading threads and the comparison logic. Each queue holds buffers read from its corresponding file.
- **Buffer Pool:** Continues utilizing the `BufferPool` for managing byte arrays to minimize memory allocation.
- **Loader Method:** A new private method `loader` is introduced to encapsulate the file reading process within each virtual thread, placing read data into the respective queue.
- **Error Handling:** Improved exception handling to properly cancel tasks and propagate exceptions. Interruptions are handled correctly.

These changes should result in significantly improved throughput when comparing large files, especially on systems with multiple cores.


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

src/checkDataIntegrity/BufferPool.java

@@ -4,18 +4,26 @@ import java.util.concurrent.LinkedBlockingQueue;
 
 class BufferPool {
 	
-	private static final LinkedBlockingQueue<byte[]> queue = new LinkedBlockingQueue<>();
+	private static final LinkedBlockingQueue<Buffer> queue = new LinkedBlockingQueue<>();
 	
-	static byte[] get() {
+	static Buffer get() {
-		byte[] buf = queue.poll();
+		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;
+		
+	}
+	
 }

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

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<Path> {
 	
 	private final Args args;
 	
 	private final ProgramState state;
 	
+	private final ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();
+	
+	private final BlockingQueue<Buffer> queue0 = new LinkedBlockingQueue<>(1);
+	
+	private final BlockingQueue<Buffer> 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<Path> {
 			final String rel = path.toString().substring(this.args.path0().length());
 			final Path p1 = Path.of(this.args.path1() + rel);
 			
-			int len0, len1;
+			try (RandomAccessFile raf0 = new RandomAccessFile(path.toFile(), "r"); RandomAccessFile raf1 = new RandomAccessFile(p1.toFile(), "r")) {
-			final byte[] buf0 = BufferPool.get();
+				final Future<?> future0 = this.executorService.submit(() -> this.loader(raf0, this.queue0));
-			final byte[] buf1 = BufferPool.get();
+				final Future<?> future1 = this.executorService.submit(() -> this.loader(raf1, this.queue1));
-			try (
+				try {
-					RandomAccessFile raf0 = new RandomAccessFile(path.toFile(), "r");
+					while (true) {
-					RandomAccessFile raf1 = new RandomAccessFile(p1.toFile(), "r")) {
+						final Buffer buf0 = this.queue0.take();
-				
+						final Buffer buf1 = this.queue1.take();
-				do {
+						try {
-					len0 = raf0.read(buf0);
+							if (buf0.len != buf1.len) {
-					len1 = raf1.read(buf1);
+								System.out.println(rel + ": Length mismatch");
-					if (len0 != len1) {
+								this.state.countErr().incrementAndGet();
-						System.out.println(rel + ": Length mismatch");
+								break;
-						this.state.countErr().incrementAndGet();
+							}
-						break;
+							if (buf0.len == -1) {
-					}
+								break;
-					if (len0 > -1) {
+							}
-						this.state.bytes().addAndGet(len0);
+							this.state.bytes().addAndGet(buf0.len);
-						if (!Arrays.equals(buf0, 0, len0, buf1, 0, len1)) {
+							if (!Arrays.equals(buf0.buf, 0, buf0.len, buf1.buf, 0, buf1.len)) {
-							System.out.println(rel + ": Data mismatch");
+								System.out.println(rel + ": Data mismatch");
-							this.state.countErr().incrementAndGet();
+								this.state.countErr().incrementAndGet();
-							break;
+								break;
+							}
+						}
+						finally {
+							BufferPool.put(buf0);
+							BufferPool.put(buf1);
 						}
 					}
-				} while (len0 > -1);
+				}
-			}
+				catch (final Exception ex) {
-			finally {
+					future0.cancel(true);
-				BufferPool.put(buf0);
+					future1.cancel(true);
-				BufferPool.put(buf1);
+					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<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) {
+		}
+	}
+	
 }