Merge pull request #2234 from jplag/feature/parallel-process

Further parallelization of the JPlag pipeline

Author: tsaglam

core/src/main/java/de/jplag/SubmissionSet.java

@@ -4,6 +4,10 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
@@ -39,11 +43,11 @@ public class SubmissionSet {
     private final Submission baseCodeSubmission;
 
     private final JPlagOptions options;
-    private int errors = 0;
-    private String currentSubmissionName;
+    private final AtomicInteger errors = new AtomicInteger(0);
 
     /**
-     * @param submissions Submissions to check for plagiarism.
+     * Creates a submissions set and parses all submissions.
+     * @param submissions list of submissions to check for plagiarism.
      * @param baseCode Base code submission if it exists or {@code null}.
      * @param options The JPlag options
      * @throws ExitException if the submissions cannot be parsed.
@@ -52,7 +56,10 @@ public SubmissionSet(List<Submission> submissions, Submission baseCode, JPlagOpt
         this.allSubmissions = submissions;
         this.baseCodeSubmission = baseCode;
         this.options = options;
-        parseAllSubmissions();
+        parseSubmissions(allSubmissions);
+        if (baseCodeSubmission != null) {
+            parseBaseCodeSubmission(baseCodeSubmission);
+        }
         this.submissions = filterValidSubmissions();
         invalidSubmissions = filterInvalidSubmissions();
     }
@@ -111,7 +118,12 @@ public void normalizeSubmissions() {
         if (baseCodeSubmission != null) {
             baseCodeSubmission.normalize();
         }
-        ProgressBarLogger.iterate(ProgressBarType.TOKEN_STRING_NORMALIZATION, submissions, Submission::normalize);
+        ProgressBar progressBar = ProgressBarLogger.createProgressBar(ProgressBarType.TOKEN_SEQUENCE_NORMALIZATION, submissions.size());
+        submissions.parallelStream().forEach(submission -> {
+            submission.normalize();
+            progressBar.step();
+        });
+        progressBar.dispose();
     }
 
     private List<Submission> filterValidSubmissions() {
@@ -122,17 +134,6 @@ private List<Submission> filterInvalidSubmissions() {
         return allSubmissions.stream().filter(it -> it.getState() != VALID).toList();
     }
 
-    private void parseAllSubmissions() throws ExitException {
-        try {
-            parseSubmissions(allSubmissions);
-            if (baseCodeSubmission != null) {
-                parseBaseCodeSubmission(baseCodeSubmission);
-            }
-        } catch (OutOfMemoryError exception) {
-            throw new SubmissionException("Out of memory during parsing of submission \"" + currentSubmissionName + "\"", exception);
-        }
-    }
-
     /**
      * Parse the given base code submission.
      */
@@ -152,30 +153,54 @@ private void parseBaseCodeSubmission(Submission baseCode) throws BasecodeExcepti
      * Parse all given submissions.
      * @param submissions The list of submissions
      */
-    private void parseSubmissions(List<Submission> submissions) throws LanguageException {
+    private void parseSubmissions(List<Submission> submissions) throws ExitException {
         if (submissions.isEmpty()) {
             logger.error("No submissions to parse!");
             return;
         }
 
         ProgressBar progressBar = ProgressBarLogger.createProgressBar(ProgressBarType.PARSING, submissions.size());
-        for (Submission submission : submissions) {
 
-            logger.trace("------ Parsing submission: " + submission.getName());
-            currentSubmissionName = submission.getName();
-
-            boolean successful = submission.parse(options.debugParser(), options.normalize(), options.minimumTokenMatch());
-            if (!successful) {
-                errors++;
-                logger.debug("ERROR -> Submission {} removed with reason {}", currentSubmissionName, submission.getState());
+        if (options.language().expectsSubmissionOrder()) {
+            for (Submission submission : submissions) {
+                parseSingleSubmission(progressBar, submission);
             }
-            progressBar.step();
+        } else {
+            parseSubmissionsInParallel(submissions, progressBar);
         }
+
         progressBar.dispose();
 
-        int validSubmissions = submissions.size() - errors;
+        int validSubmissions = submissions.size() - errors.get();
         logger.debug("{} submissions parsed successfully!", validSubmissions);
-        logger.debug("{} parser error{}!", errors, errors != 1 ? "s" : "");
+        logger.debug("{} parser error{}!", errors, errors.get() != 1 ? "s" : "");
+    }
+
+    private void parseSubmissionsInParallel(List<Submission> submissions, ProgressBar progressBar) throws SubmissionException {
+        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
+            for (Submission submission : submissions) {
+                executor.submit(() -> {
+                    parseSingleSubmission(progressBar, submission);
+                    return null; // Ensure the lambda is a Callable for exception handling
+                });
+            }
+            executor.shutdown();
+            executor.awaitTermination(24, TimeUnit.HOURS); // Maximum time all processing can take.
+        } catch (InterruptedException exception) {
+            throw new SubmissionException("Error while parsing the submissions.", exception);
+        }
+    }
+
+    /**
+     * Parses a single submission (thread safe).
+     */
+    private void parseSingleSubmission(ProgressBar progressBar, Submission submission) throws LanguageException {
+        boolean successful = submission.parse(options.debugParser(), options.normalize(), options.minimumTokenMatch());
+        if (!successful) {
+            errors.incrementAndGet();
+            logger.debug("ERROR -> Submission {} removed with reason {}", submission.getName(), submission.getState());
+        }
+        progressBar.step();
     }
 
 }

core/src/main/java/de/jplag/logging/ProgressBarType.java

@@ -4,13 +4,13 @@
  * The available processes. Used as a hint for the ui, which step JPlag is currently performing.
  */
 public enum ProgressBarType {
-    LOADING("Loading Submissions  ", false),
-    PARSING("Parsing Submissions  ", false),
+    LOADING("Loading Submissions", false),
+    PARSING("Parsing Submissions", false),
     TOKEN_VALUE_CREATION("Preparing Submissions", false),
-    COMPARING("Comparing Submissions", false),
-    MATCH_MERGING("Merging matched subsequences ", false),
-    TOKEN_STRING_NORMALIZATION("Normalizing Token Sequence", false),
-    CLUSTERING("Finding clusters ", true);
+    COMPARING("Comparing Submission Pairs", false),
+    MATCH_MERGING("Merging Matched Subsequences ", false),
+    TOKEN_SEQUENCE_NORMALIZATION("Normalizing Token Sequences", false),
+    CLUSTERING("Finding Clusters ", true);
 
     private final String defaultText;
     private final boolean isIdleBar;

core/src/main/java/de/jplag/merging/MatchMerging.java

@@ -10,6 +10,7 @@
 import de.jplag.SharedTokenType;
 import de.jplag.Submission;
 import de.jplag.Token;
+import de.jplag.logging.ProgressBar;
 import de.jplag.logging.ProgressBarLogger;
 import de.jplag.logging.ProgressBarType;
 import de.jplag.options.JPlagOptions;
@@ -43,29 +44,31 @@ public MatchMerging(JPlagOptions options) {
      */
     public JPlagResult mergeMatchesOf(JPlagResult result) {
         long timeBeforeStartInMillis = System.currentTimeMillis();
-
         List<JPlagComparison> comparisons = new ArrayList<>(result.getAllComparisons());
-        List<JPlagComparison> comparisonsMerged = new ArrayList<>();
-
-        ProgressBarLogger.iterate(ProgressBarType.MATCH_MERGING, comparisons, comparison -> {
-            numberOfMerges = 0;
-            Submission leftSubmission = comparison.firstSubmission().copy();
-            Submission rightSubmission = comparison.secondSubmission().copy();
-            List<Match> globalMatches = new ArrayList<>(comparison.matches());
-            globalMatches.addAll(comparison.ignoredMatches());
-            globalMatches = mergeNeighbors(globalMatches, leftSubmission, rightSubmission);
-            globalMatches = globalMatches.stream().filter(it -> it.length() >= options.minimumTokenMatch()).toList();
-            if (numberOfMerges >= options.mergingOptions().minimumRequiredMerges()) {
-                comparisonsMerged.add(new JPlagComparison(leftSubmission, rightSubmission, globalMatches, new ArrayList<>()));
-            } else {
-                comparisonsMerged.add(comparison);
-            }
-        });
+
+        ProgressBar progressBar = ProgressBarLogger.createProgressBar(ProgressBarType.MATCH_MERGING, comparisons.size());
+        List<JPlagComparison> comparisonsMerged = comparisons.parallelStream().map(it -> mergeMatchesOf(it, progressBar)).toList();
+        progressBar.dispose();
 
         long durationInMillis = System.currentTimeMillis() - timeBeforeStartInMillis;
         return new JPlagResult(comparisonsMerged, result.getSubmissions(), result.getDuration() + durationInMillis, options);
     }
 
+    private JPlagComparison mergeMatchesOf(JPlagComparison comparison, ProgressBar progressBar) {
+        numberOfMerges = 0;
+        Submission leftSubmission = comparison.firstSubmission().copy();
+        Submission rightSubmission = comparison.secondSubmission().copy();
+        List<Match> globalMatches = new ArrayList<>(comparison.matches());
+        globalMatches.addAll(comparison.ignoredMatches());
+        globalMatches = mergeNeighbors(globalMatches, leftSubmission, rightSubmission);
+        globalMatches = globalMatches.stream().filter(it -> it.length() >= options.minimumTokenMatch()).toList();
+        progressBar.step();
+        if (numberOfMerges >= options.mergingOptions().minimumRequiredMerges()) {
+            return new JPlagComparison(leftSubmission, rightSubmission, globalMatches, new ArrayList<>());
+        }
+        return comparison;
+    }
+
     /**
      * Computes neighbors by sorting based on order of matches in the left and right submissions and then checking which are
      * next to each other in both.

language-antlr-utils/src/main/java/de/jplag/antlr/AbstractAntlrLanguage.java

@@ -1,66 +1,21 @@
 package de.jplag.antlr;
 
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-import java.io.File;
 import java.util.Set;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import de.jplag.ParsingException;
 import de.jplag.antlr.testLanguage.TestLanguage;
-import de.jplag.antlr.testLanguage.TestParserAdapter;
 
 /**
  * Some tests for the abstract antlr language
  */
 class LanguageTest {
-    @Test
-    void testExceptionForNoDefinedParser() {
-        LanguageWithoutParser lang = new LanguageWithoutParser();
-        Set<File> emptySet = Set.of();
-        assertThrows(UnsupportedOperationException.class, () -> lang.parse(emptySet, false));
-    }
 
     @Test
     void testLanguageWithStaticParser() throws ParsingException {
         TestLanguage lang = new TestLanguage();
         Assertions.assertEquals(0, lang.parse(Set.of(), false).size());
     }
-
-    @Test
-    void testLanguageWithLazyParser() throws ParsingException {
-        LanguageWithLazyParser lang = new LanguageWithLazyParser();
-        Assertions.assertEquals(0, lang.parse(Set.of(), false).size());
-    }
-
-    private static class LanguageWithoutParser extends AbstractAntlrLanguage {
-        @Override
-        public String[] suffixes() {
-            return new String[0];
-        }
-
-        @Override
-        public String getName() {
-            return null;
-        }
-
-        @Override
-        public String getIdentifier() {
-            return null;
-        }
-
-        @Override
-        public int minimumTokenMatch() {
-            return 0;
-        }
-    }
-
-    private static class LanguageWithLazyParser extends LanguageWithoutParser {
-        @Override
-        protected AbstractAntlrParserAdapter<?> initializeParser(boolean normalize) {
-            return new TestParserAdapter();
-        }
-    }
 }

language-antlr-utils/src/test/java/de/jplag/antlr/LanguageTest.java

@@ -1,14 +1,14 @@
 package de.jplag.antlr.testLanguage;
 
-import de.jplag.antlr.AbstractAntlrLanguage;
-
-public class TestLanguage extends AbstractAntlrLanguage {
-    /**
-     * New instance
-     */
-    public TestLanguage() {
-        super(new TestParserAdapter());
-    }
+import java.io.File;
+import java.util.List;
+import java.util.Set;
+
+import de.jplag.Language;
+import de.jplag.ParsingException;
+import de.jplag.Token;
+
+public class TestLanguage implements Language {
 
     @Override
     public String[] suffixes() {
@@ -29,4 +29,9 @@ public String getIdentifier() {
     public int minimumTokenMatch() {
         return 8;
     }
+
+    @Override
+    public List<Token> parse(Set<File> files, boolean normalize) throws ParsingException {
+        return new TestParserAdapter().parse(files);
+    }
 }

language-antlr-utils/src/test/java/de/jplag/antlr/testLanguage/TestLanguage.java

@@ -15,12 +15,6 @@ public class CLanguage implements Language {
     private static final String NAME = "C";
     private static final String IDENTIFIER = "c";
 
-    private final Scanner scanner; // c code is scanned not parsed
-
-    public CLanguage() {
-        scanner = new Scanner();
-    }
-
     @Override
     public String[] suffixes() {
         return new String[] {".cpp", ".CPP", ".cxx", ".CXX", ".c++", ".C++", ".c", ".C", ".cc", ".CC", ".h", ".H", ".hpp", ".HPP", ".hh", ".HH"};
@@ -43,6 +37,6 @@ public int minimumTokenMatch() {
 
     @Override
     public List<Token> parse(Set<File> files, boolean normalize) throws ParsingException {
-        return this.scanner.scan(files);
+        return new Scanner().scan(files);
     }
 }