report duration of PendingFileCompleter (#2947)

Author: Vladimir Kotal

opengrok-indexer/pom.xml

@@ -46,6 +46,11 @@ Portions Copyright (c) 2017-2018, Chris Fraire <cfraire@me.com>.
             <groupId>org.apache.bcel</groupId>
             <artifactId>bcel</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>${apache-commons-lang3.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.lucene</groupId>
             <artifactId>lucene-core</artifactId>

opengrok-indexer/src/main/java/org/opengrok/indexer/index/IndexDatabase.java

@@ -99,6 +99,7 @@
 import org.opengrok.indexer.util.ForbiddenSymlinkException;
 import org.opengrok.indexer.util.IOUtils;
 import org.opengrok.indexer.util.ObjectPool;
+import org.opengrok.indexer.util.Progress;
 import org.opengrok.indexer.util.Statistics;
 import org.opengrok.indexer.util.TandemPath;
 import org.opengrok.indexer.web.Util;
@@ -1026,27 +1027,6 @@ private boolean isLocal(String path) {
         return local;
     }
 
-    private void printProgress(String dir, int currentCount, int totalCount) {
-        if (totalCount > 0 && RuntimeEnvironment.getInstance().isPrintProgress()) {
-            Level currentLevel;
-            if (currentCount <= 1 || currentCount >= totalCount ||
-                    currentCount % 100 == 0) {
-                currentLevel = Level.INFO;
-            } else if (currentCount % 50 == 0) {
-                currentLevel = Level.FINE;
-            } else if (currentCount % 10 == 0) {
-                currentLevel = Level.FINER;
-            } else {
-                currentLevel = Level.FINEST;
-            }
-            if (LOGGER.isLoggable(currentLevel)) {
-                LOGGER.log(currentLevel, "Progress: {0} ({1}%) for {2}",
-                        new Object[]{currentCount, currentCount * 100.0f /
-                                totalCount, dir});
-            }
-        }
-    }
-
     /**
      * Executes the first, serial stage of indexing, recursively.
      * <p>Files at least are counted, and any deleted or updated files (based on
@@ -1180,7 +1160,7 @@ private void indexParallel(String dir, IndexDownArgs args) {
         ObjectPool<Ctags> ctagsPool = parallelizer.getCtagsPool();
 
         Map<Boolean, List<IndexFileWork>> bySuccess = null;
-        try {
+        try (Progress progress = new Progress(LOGGER, dir, worksCount)) {
             bySuccess = parallelizer.getForkJoinPool().submit(() ->
                 args.works.parallelStream().collect(
                 Collectors.groupingByConcurrent((x) -> {
@@ -1225,8 +1205,7 @@ private void indexParallel(String dir, IndexDownArgs args) {
                             }
                         }
 
-                        int ncount = currentCounter.incrementAndGet();
-                        printProgress(dir, ncount, worksCount);
+                        progress.increment();
                         return ret;
                     }
                 }))).get();

opengrok-indexer/src/main/java/org/opengrok/indexer/index/PendingFileCompleter.java

@@ -34,6 +34,8 @@
 import java.nio.file.SimpleFileVisitor;
 import java.nio.file.StandardCopyOption;
 import java.nio.file.attribute.BasicFileAttributes;
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
@@ -43,7 +45,10 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
+
+import org.apache.commons.lang3.time.DurationFormatUtils;
 import org.opengrok.indexer.logger.LoggerFactory;
+import org.opengrok.indexer.util.Progress;
 import org.opengrok.indexer.util.TandemPath;
 
 /**
@@ -166,12 +171,27 @@ public boolean add(PendingFileRenaming e) {
      * @throws java.io.IOException if an I/O error occurs
      */
     public int complete() throws IOException {
+        Instant start = Instant.now();
         int numDeletions = completeDeletions();
-        LOGGER.log(Level.FINE, "deleted {0} file(s)", numDeletions);
+        LOGGER.log(Level.FINE, "deleted {0} file(s) (took {1})",
+                new Object[]{numDeletions, DurationFormatUtils.
+                        formatDurationWords(Duration.between(start, Instant.now()).toMillis(),
+                        true, true)});
+
+        start = Instant.now();
         int numRenamings = completeRenamings();
-        LOGGER.log(Level.FINE, "renamed {0} file(s)", numRenamings);
+        LOGGER.log(Level.FINE, "renamed {0} file(s) (took {1})",
+                new Object[]{numRenamings, DurationFormatUtils.
+                        formatDurationWords(Duration.between(start, Instant.now()).toMillis(),
+                true, true)});
+
+        start = Instant.now();
         int numLinkages = completeLinkages();
-        LOGGER.log(Level.FINE, "affirmed links for {0} path(s)", numLinkages);
+        LOGGER.log(Level.FINE, "affirmed links for {0} path(s) (took {1})",
+                new Object[]{numLinkages, DurationFormatUtils.
+                        formatDurationWords(Duration.between(start, Instant.now()).toMillis(),
+                        true, true)});
+
         return numDeletions + numRenamings + numLinkages;
     }
 
@@ -194,18 +214,21 @@ private int completeRenamings() throws IOException {
             new PendingFileRenamingExec(f.getTransientPath(),
                 f.getAbsolutePath())).collect(
             Collectors.toList());
-
-        Map<Boolean, List<PendingFileRenamingExec>> bySuccess =
-            pendingExecs.parallelStream().collect(
-            Collectors.groupingByConcurrent((x) -> {
-                try {
-                    doRename(x);
-                    return true;
-                } catch (IOException e) {
-                    x.exception = e;
-                    return false;
-                }
-            }));
+        Map<Boolean, List<PendingFileRenamingExec>> bySuccess;
+
+        try (Progress progress = new Progress(LOGGER, "pending renames", numPending)) {
+            bySuccess = pendingExecs.parallelStream().collect(
+                            Collectors.groupingByConcurrent((x) -> {
+                                progress.increment();
+                                try {
+                                    doRename(x);
+                                    return true;
+                                } catch (IOException e) {
+                                    x.exception = e;
+                                    return false;
+                                }
+                            }));
+        }
         renames.clear();
 
         List<PendingFileRenamingExec> failures = bySuccess.getOrDefault(
@@ -240,18 +263,21 @@ private int completeDeletions() throws IOException {
             parallelStream().map(f ->
             new PendingFileDeletionExec(f.getAbsolutePath())).collect(
             Collectors.toList());
-
-        Map<Boolean, List<PendingFileDeletionExec>> bySuccess =
-            pendingExecs.parallelStream().collect(
-            Collectors.groupingByConcurrent((x) -> {
-                try {
-                    doDelete(x);
-                    return true;
-                } catch (IOException e) {
-                    x.exception = e;
-                    return false;
-                }
-            }));
+        Map<Boolean, List<PendingFileDeletionExec>> bySuccess;
+
+        try (Progress progress = new Progress(LOGGER, "pending deletions", numPending)) {
+            bySuccess = pendingExecs.parallelStream().collect(
+                            Collectors.groupingByConcurrent((x) -> {
+                                progress.increment();
+                                try {
+                                    doDelete(x);
+                                    return true;
+                                } catch (IOException e) {
+                                    x.exception = e;
+                                    return false;
+                                }
+                            }));
+        }
         deletions.clear();
 
         List<PendingFileDeletionExec> successes = bySuccess.getOrDefault(
@@ -292,17 +318,20 @@ private int completeLinkages() throws IOException {
                 new PendingSymlinkageExec(f.getSourcePath(),
                         f.getTargetRelPath())).collect(Collectors.toList());
 
-        Map<Boolean, List<PendingSymlinkageExec>> bySuccess =
-            pendingExecs.parallelStream().collect(
-                Collectors.groupingByConcurrent((x) -> {
-                    try {
-                        doLink(x);
-                        return true;
-                    } catch (IOException e) {
-                        x.exception = e;
-                        return false;
-                    }
-                }));
+        Map<Boolean, List<PendingSymlinkageExec>> bySuccess;
+        try (Progress progress = new Progress(LOGGER, "pending renames", numPending)) {
+            bySuccess = pendingExecs.parallelStream().collect(
+                            Collectors.groupingByConcurrent((x) -> {
+                                progress.increment();
+                                try {
+                                    doLink(x);
+                                    return true;
+                                } catch (IOException e) {
+                                    x.exception = e;
+                                    return false;
+                                }
+                            }));
+        }
         linkages.clear();
 
         List<PendingSymlinkageExec> failures = bySuccess.getOrDefault(

opengrok-indexer/src/main/java/org/opengrok/indexer/util/Progress.java

@@ -0,0 +1,146 @@
+/*
+ * CDDL HEADER START
+ *
+ * The contents of this file are subject to the terms of the
+ * Common Development and Distribution License (the "License").
+ * You may not use this file except in compliance with the License.
+ *
+ * See LICENSE.txt included in this distribution for the specific
+ * language governing permissions and limitations under the License.
+ *
+ * When distributing Covered Code, include this CDDL HEADER in each
+ * file and include the License file at LICENSE.txt.
+ * If applicable, add the following below this CDDL HEADER, with the
+ * fields enclosed by brackets "[]" replaced with your own identifying
+ * information: Portions Copyright [yyyy] [name of copyright owner]
+ *
+ * CDDL HEADER END
+ */
+
+/*
+ * Copyright (c) 2007, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Portions Copyright (c) 2017-2018, Chris Fraire <cfraire@me.com>.
+ */
+
+package org.opengrok.indexer.util;
+
+import org.opengrok.indexer.configuration.RuntimeEnvironment;
+
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class Progress implements AutoCloseable {
+    private final Logger logger;
+    private final long totalCount;
+    private final String suffix;
+
+    private AtomicLong currentCount = new AtomicLong();
+    private Thread loggerThread = null;
+    private volatile boolean run;
+
+    private final Object sync = new Object();
+
+    /**
+     * @param logger logger instance
+     * @param suffix string suffix to identify the operation
+     * @param totalCount total count
+     */
+    public Progress(Logger logger, String suffix, long totalCount) {
+        this.logger = logger;
+        this.suffix = suffix;
+        this.totalCount = totalCount;
+
+        // Assuming printProgress configuration setting cannot be changed on the fly.
+        if (totalCount > 0 && RuntimeEnvironment.getInstance().isPrintProgress()) {
+            // spawn a logger thread.
+            run = true;
+            loggerThread = new Thread(this::logLoop,
+                    "progress-thread-" + suffix.replaceAll(" ", "_"));
+            loggerThread.start();
+        }
+    }
+
+    // for testing
+    Thread getLoggerThread() {
+        return loggerThread;
+    }
+
+    /**
+     * Increment counter. The actual logging will be done eventually.
+     */
+    public void increment() {
+        this.currentCount.incrementAndGet();
+
+        if (loggerThread != null) {
+            // nag the thread.
+            synchronized (sync) {
+                sync.notify();
+            }
+        }
+    }
+
+    private void logLoop() {
+        long cachedCount = 0;
+
+        while (true) {
+            long currentCount = this.currentCount.get();
+            Level currentLevel;
+
+            if (cachedCount < currentCount) {
+                if (currentCount <= 1 || currentCount % 100 == 0) {
+                    currentLevel = Level.INFO;
+                } else if (currentCount % 50 == 0) {
+                    currentLevel = Level.FINE;
+                } else if (currentCount % 10 == 0) {
+                    currentLevel = Level.FINER;
+                } else {
+                    currentLevel = Level.FINEST;
+                }
+
+                // Do not log if there was no progress.
+                if (logger.isLoggable(currentLevel)) {
+                    logger.log(currentLevel, "Progress: {0} ({1}%) for {2}",
+                            new Object[]{currentCount, currentCount * 100.0f /
+                                    totalCount, suffix});
+                }
+            }
+
+            if (!run) {
+                return;
+            }
+
+            cachedCount = currentCount;
+
+            // wait for event
+            try {
+                synchronized (sync) {
+                    if (!run) {
+                        // Loop once more to do the final logging.
+                        continue;
+                    }
+                    sync.wait();
+                }
+            } catch (InterruptedException e) {
+                logger.log(Level.WARNING, "logger thread interrupted");
+            }
+        }
+    }
+
+    @Override
+    public void close() {
+        if (loggerThread == null) {
+            return;
+        }
+
+        try {
+            run = false;
+            synchronized (sync) {
+                sync.notify();
+            }
+            loggerThread.join();
+        } catch (InterruptedException e) {
+            logger.log(Level.WARNING, "logger thread interrupted");
+        }
+    }
+}