Merge pull request #1054 from knutwalker/wcc-logging

Make sure to log progress if it is incremented by values other than 1

Author: DarthMax

core/src/main/java/org/neo4j/graphalgo/core/utils/BatchingProgressLogger.java

@@ -22,7 +22,6 @@
 import org.apache.commons.lang3.mutable.MutableLong;
 import org.neo4j.logging.Log;
 
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Supplier;
 
@@ -65,16 +64,16 @@ public BatchingProgressLogger(Log log, long taskVolume, long batchSize, String t
     @Override
     public void logProgress(Supplier<String> msgFactory) {
         progressCounter.increment();
-        long localProgress = callCounter.get().incrementAndGet();
-
-        if ((localProgress & (batchSize - 1)) == 0) {
+        var localProgress = callCounter.get();
+        if (localProgress.longValue() < batchSize && (localProgress.incrementAndGet() >= batchSize)) {
             String message = msgFactory != ProgressLogger.NO_MESSAGE ? msgFactory.get() : null;
             int percent = (int) ((progressCounter.sum() / (double) taskVolume) * 100);
             if (message == null || message.isEmpty()) {
                 log.info("[%s] %s %d%%", Thread.currentThread().getName(), task, percent);
             } else {
                 log.info("[%s] %s %d%% %s", Thread.currentThread().getName(), task, percent, message);
             }
+            localProgress.setValue(0L);
         }
     }
 
@@ -84,11 +83,11 @@ public void logProgress(long progress, Supplier<String> msgFactory) {
             return;
         }
         progressCounter.add(progress);
-        long localProgress = callCounter.get().incrementAndGet();
-
-        if ((localProgress & (batchSize -1)) == 0) {
+        var localProgress = callCounter.get();
+        if (localProgress.longValue() < batchSize && (localProgress.addAndGet(progress) >= batchSize)) {
             int percent = (int) ((progressCounter.sum() / (double) taskVolume) * 100);
             log.info("[%s] %s %d%%", Thread.currentThread().getName(), task, percent);
+            localProgress.setValue(localProgress.longValue() & (batchSize - 1));
         }
     }
 

core/src/test/java/org/neo4j/graphalgo/core/utils/BatchingProgressLoggerTest.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright (c) 2017-2020 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.graphalgo.core.utils;
+
+import org.eclipse.collections.impl.utility.ListIterate;
+import org.junit.jupiter.api.Test;
+import org.neo4j.graphalgo.TestLog;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.neo4j.graphalgo.utils.StringFormatting.formatWithLocale;
+
+class BatchingProgressLoggerTest {
+
+    @Test
+    void mustLogProgressOnlyAfterBatchSizeInvocations() {
+        var log = new TestLog();
+        var taskVolume = 42;
+        var batchSize = 8;
+        var logger = new BatchingProgressLogger(
+            log,
+            taskVolume,
+            batchSize,
+            "foo",
+            /* concurrency */0
+        );
+
+        for (int i = 0; i < taskVolume; i++) {
+            int currentProgress = i;
+            logger.logProgress(() -> String.valueOf(currentProgress));
+        }
+
+        var threadName = Thread.currentThread().getName();
+        var messageTemplate = "[%s] foo %d%% %d";
+        var expectedMessages = List.of(
+            formatWithLocale(messageTemplate, threadName, 1 * batchSize * 100 / taskVolume, 1 * batchSize - 1),
+            formatWithLocale(messageTemplate, threadName, 2 * batchSize * 100 / taskVolume, 2 * batchSize - 1),
+            formatWithLocale(messageTemplate, threadName, 3 * batchSize * 100 / taskVolume, 3 * batchSize - 1),
+            formatWithLocale(messageTemplate, threadName, 4 * batchSize * 100 / taskVolume, 4 * batchSize - 1),
+            formatWithLocale(messageTemplate, threadName, 5 * batchSize * 100 / taskVolume, 5 * batchSize - 1)
+        );
+
+        var messages = log.getMessages("info");
+        assertEquals(expectedMessages, messages);
+    }
+
+    @Test
+    void mustLogProgressOnlyAfterHittingOrExceedingBatchSize() {
+        var log = new TestLog();
+        var taskVolume = 1337;
+        var progressStep = 5;
+        var batchSize = 16;
+        var logger = new BatchingProgressLogger(
+            log,
+            taskVolume,
+            batchSize,
+            "foo",
+            /* concurrency */1
+        );
+
+        for (int i = 0; i < taskVolume; i += progressStep) {
+            int currentProgress = i;
+            logger.logProgress(progressStep, () -> String.valueOf(currentProgress));
+        }
+
+        var threadName = Thread.currentThread().getName();
+        var messageTemplate = "[%s] foo %d%%";
+
+        var progressSteps = IntStream
+            .iterate(0, i -> i < taskVolume, i -> i + progressStep)
+            .boxed()
+            .collect(Collectors.toList());
+        var loggedProgressSteps = ListIterate.distinctBy(progressSteps, i -> i / batchSize);
+
+        var expectedMessages = loggedProgressSteps.stream()
+            .skip(1)
+            .map(i -> formatWithLocale(messageTemplate, threadName, i * 100 / taskVolume))
+            .collect(Collectors.toList());
+
+        var messages = log.getMessages("info");
+        assertEquals(expectedMessages, messages);
+    }
+}