De-recursify hasNext() in TextFilePLL, optimizing sync change data read

wetneb · Jan 27, 2024 · d56475b · d56475b
1 parent f8a3302
commit d56475b
Show file tree

Hide file tree

Showing 2 changed files with 48 additions and 41 deletions.
diff --git a/modules/refine-local-runner/src/main/java/org/openrefine/runners/local/pll/TextFilePLL.java b/modules/refine-local-runner/src/main/java/org/openrefine/runners/local/pll/TextFilePLL.java
@@ -250,53 +250,59 @@ protected CloseableIterator<String> compute(Partition partition, IterationContex
 
                 @Override
                 public boolean hasNext() {
-                    long currentPosition = textPartition.start + (countingIs == null ? 0 : countingIs.getCount());
-                    try {
-                        if (!nextLineAttempted && ((currentPosition <= textPartition.getEnd() || textPartition.getEnd() < 0) || synchronous)) {
-                            if (synchronous) {
-                                lineNumberReader.mark(4096);
-                                // TODO add logic to bump this readAheadLimit (restart from the beginning of the
-                                // stream…)
+                    while (!nextLineAttempted && nextLine == null && !endMarkerFound) {
+                        long currentPosition = textPartition.start + (countingIs == null ? 0 : countingIs.getCount());
+                        try {
+                            if (!nextLineAttempted && ((currentPosition <= textPartition.getEnd() || textPartition.getEnd() < 0) || synchronous)) {
+                                if (synchronous) {
+                                    lineNumberReader.mark(4096);
+                                    // TODO add logic to bump this readAheadLimit (restart from the beginning of the
+                                    // stream…)
+                                }
+                                nextLineAttempted = true;
+                                if (lineNumberReader != null) {
+                                    nextLine = lineNumberReader.readLine();
+                                } else {
+                                    nextLine = lineReader.readLine();
+                                }
+                                lastOffsetSeen = currentPosition;
+                                if (endMarker != null && nextLine != null && nextLine.startsWith(endMarker)) {
+                                    endMarkerFound = true;
+                                    nextLine = null;
+                                }
                             }
-                            if (lineNumberReader != null) {
-                                nextLine = lineNumberReader.readLine();
-                            } else {
-                                nextLine = lineReader.readLine();
+                            if (nextLine == null && lastOffsetSeen > lastOffsetReported) {
+                                reportProgress();
                             }
-                            nextLineAttempted = true;
-                            lastOffsetSeen = currentPosition;
-                            if (endMarker != null && nextLine != null && nextLine.startsWith(endMarker)) {
-                                endMarkerFound = true;
+                        } catch (EOFException | ZipException e) {
+                            if (ignoreEarlyEOF) {
                                 nextLine = null;
+                            } else {
+                                throw new UncheckedIOException(e);
                             }
-                        }
-                        if (nextLine == null && lastOffsetSeen > lastOffsetReported) {
-                            reportProgress();
-                        }
-                    } catch (EOFException | ZipException e) {
-                        if (ignoreEarlyEOF) {
-                            nextLine = null;
-                        } else {
+                        } catch (IOException e) {
                             throw new UncheckedIOException(e);
                         }
-                    } catch (IOException e) {
-                        throw new UncheckedIOException(e);
-                    }
-                    if (nextLineAttempted && nextLine == null && synchronous && !endMarkerFound) {
-                        try {
-                            lineNumberReader.reset();
-                            if (watchService == null) {
-                                watchService = FileSystems.getDefault().newWatchService();
+                        if (nextLine == null && synchronous && !endMarkerFound) {
+                            try {
+                                lineNumberReader.reset();
+                                if (watchService == null) {
+                                    watchService = FileSystems.getDefault().newWatchService();
+                                    Path pathToWatch = Paths.get(textPartition.getPath().getParent());
+                                    pathToWatch.register(watchService, ENTRY_MODIFY);
+                                }
+                                WatchKey key = watchService.poll(1000, TimeUnit.MILLISECONDS);
+                                if (key != null) {
+                                    key.reset();
+                                }
+                                nextLineAttempted = false;
+                            } catch (IOException e) {
+                                throw new UncheckedIOException(e);
+                            } catch (InterruptedException e) {
+                                Thread.currentThread().interrupt();
                             }
-                            Path pathToWatch = Paths.get(textPartition.getPath().getParent());
-                            pathToWatch.register(watchService, ENTRY_MODIFY);
-                            WatchKey key = watchService.poll(1000, TimeUnit.MILLISECONDS);
-                            nextLineAttempted = false;
-                            return hasNext();
-                        } catch (IOException e) {
-                            throw new UncheckedIOException(e);
-                        } catch (InterruptedException e) {
-                            Thread.currentThread().interrupt();
+                        } else {
+                            nextLineAttempted = true;
                         }
                     }
                     return nextLine != null;

diff --git a/.../refine-local-runner/src/test/java/org/openrefine/runners/local/pll/TextFilePLLTests.java b/.../refine-local-runner/src/test/java/org/openrefine/runners/local/pll/TextFilePLLTests.java
@@ -359,12 +359,13 @@ public void testSynchronousRead() throws IOException, InterruptedException {
                 writer2.write("the\n");
                 writer2.write("second\n");
                 writer2.write("partition\n");
+                writer2.write("end\n");
                 writer2.flush();
 
                 // give the reader a bit of time to catch up
                 sleep(200);
                 synchronized (readStrings) {
-                    // we should have read all the lines (without the end marker)
+                    // we should have read all the lines (without the end markers)
                     assertTrue(readStrings.size() >= 11);
                     assertEquals(readStrings.get(10), "hi");
                 }