make parallelism of suggester rebuild tunable

Author: Vladimir Kotal

apiary.apib

@@ -366,13 +366,17 @@ with a colon. If the type cannot be determined "N/A" is returned.
 
 ### rebuilds suggester data for all projects [PUT]
 
++ Response 204
+
 ## Suggester project data rebuild [/suggest/rebuild/{project}]
 
 ### rebuilds suggester data for given project [PUT]
 
 + Parameters
   + project - project name
 
++ Response 204
+
 ## Index searchers refresh [/system/refresh]
 
 ### refreshes index searchers for specified project [PUT]

dev/checkstyle/suppressions.xml

@@ -13,7 +13,7 @@
 
     <suppress checks="ParameterNumber" files="CtagsReader\.java|Definitions\.java|PerlLexHelper\.java|
         |JFlexXrefUtils\.java|RubyLexHelper\.java|FileAnalyzerFactory\.java|SearchController\.java|
-        |Context\.java|HistoryContext\.java|Indexer\.java" />
+        |Context\.java|HistoryContext\.java|Indexer\.java|Suggester\.java" />
 
     <suppress checks="MethodLength" files="Indexer\.java" />
 </suppressions>

opengrok-indexer/src/main/java/org/opengrok/indexer/configuration/SuggesterConfig.java

@@ -50,6 +50,7 @@ public class SuggesterConfig {
     public static final String REBUILD_CRON_CONFIG_DEFAULT = "0 0 * * *"; // every day at midnight
     public static final int BUILD_TERMINATION_TIME_DEFAULT = 1800; // half an hour should be enough
     public static final int TIME_THRESHOLD_DEFAULT = 2000; // 2 sec
+    public static final int REBUILD_THREAD_POOL_PERCENT_NCPUS_DEFAULT = 80;
 
     public static final Set<String> allowedProjectsDefault = null;
     public static final Set<String> allowedFieldsDefault = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
@@ -135,6 +136,11 @@ public class SuggesterConfig {
      */
     private int timeThreshold;
 
+    /**
+     * Number of threads used for rebuild pool expressed in percent of available CPUs in the system.
+     */
+    private int rebuildThreadPoolSizeInNcpuPercent;
+
     public SuggesterConfig() {
         setEnabled(ENABLED_DEFAULT);
         setMaxResults(MAX_RESULTS_DEFAULT);
@@ -151,6 +157,7 @@ public SuggesterConfig() {
         // do not use setter because indexer invocation with --man will fail
         rebuildCronConfig = REBUILD_CRON_CONFIG_DEFAULT;
         setBuildTerminationTime(BUILD_TERMINATION_TIME_DEFAULT);
+        setRebuildThreadPoolSizeInNcpuPercent(REBUILD_THREAD_POOL_PERCENT_NCPUS_DEFAULT);
     }
 
     public boolean isEnabled() {
@@ -285,6 +292,17 @@ public void setTimeThreshold(final int timeThreshold) {
         this.timeThreshold = timeThreshold;
     }
 
+    public void setRebuildThreadPoolSizeInNcpuPercent(final int percent) {
+        if (percent < 0 || percent > 100) {
+            throw new IllegalArgumentException("Need percentage value");
+        }
+        this.rebuildThreadPoolSizeInNcpuPercent = percent;
+    }
+
+    public int getRebuildThreadPoolSizeInNcpuPercent() {
+        return rebuildThreadPoolSizeInNcpuPercent;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -306,14 +324,15 @@ public boolean equals(Object o) {
                 buildTerminationTime == that.buildTerminationTime &&
                 Objects.equals(allowedProjects, that.allowedProjects) &&
                 Objects.equals(allowedFields, that.allowedFields) &&
-                Objects.equals(rebuildCronConfig, that.rebuildCronConfig);
+                Objects.equals(rebuildCronConfig, that.rebuildCronConfig) &&
+                rebuildThreadPoolSizeInNcpuPercent == that.rebuildThreadPoolSizeInNcpuPercent;
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(enabled, maxResults, minChars, allowedProjects, maxProjects, allowedFields,
                 allowComplexQueries, allowMostPopular, showScores, showProjects, showTime, rebuildCronConfig,
-                buildTerminationTime);
+                buildTerminationTime, rebuildThreadPoolSizeInNcpuPercent);
     }
 
 }

opengrok-indexer/src/test/java/org/opengrok/indexer/configuration/SuggesterConfigTest.java

@@ -0,0 +1,44 @@
+/*
+ * 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) 2018 Oracle and/or its affiliates. All rights reserved.
+ */
+
+package org.opengrok.indexer.configuration;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class SuggesterConfigTest {
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testRebuildParallelismNegative() {
+        SuggesterConfig sconfig = new SuggesterConfig();
+        sconfig.setRebuildThreadPoolSizeInNcpuPercent(-1);
+    }
+
+    @Test
+    public void testRebuildParallelismPositive() {
+        SuggesterConfig sconfig = new SuggesterConfig();
+        sconfig.setRebuildThreadPoolSizeInNcpuPercent(70);
+        assertEquals(70, sconfig.getRebuildThreadPoolSizeInNcpuPercent());
+    }
+}

opengrok-web/src/main/java/org/opengrok/web/api/v1/controller/SuggesterController.java

@@ -168,15 +168,13 @@ public SuggesterConfig getConfig() {
     @PUT
     @Path("/rebuild")
     public void rebuild() {
-        // TODO run this async in a thread so that the request get a response instantly
-        suggester.rebuild();
+        new Thread(() -> suggester.rebuild()).start();
     }
 
     @PUT
     @Path("/rebuild/{project}")
     public void rebuild(@PathParam("project") final String project) {
-        // TODO run this async in a thread so that the request get a response instantly
-        suggester.rebuild(project);
+        new Thread(() -> suggester.rebuild(project)).start();
     }
 
     /**

opengrok-web/src/main/java/org/opengrok/web/api/v1/suggester/provider/service/impl/SuggesterServiceImpl.java

@@ -35,15 +35,13 @@
 import org.opengrok.suggest.Suggester.NamedIndexReader;
 import org.opengrok.suggest.Suggester.Suggestions;
 import org.opengrok.suggest.query.SuggesterQuery;
-import org.opengrok.indexer.configuration.Configuration;
 import org.opengrok.indexer.configuration.Project;
 import org.opengrok.indexer.configuration.RuntimeEnvironment;
 import org.opengrok.indexer.configuration.SuggesterConfig;
 import org.opengrok.indexer.configuration.SuperIndexSearcher;
 import org.opengrok.indexer.index.IndexDatabase;
 import org.opengrok.indexer.logger.LoggerFactory;
 import org.opengrok.web.api.v1.suggester.provider.service.SuggesterService;
-import sun.reflect.generics.reflectiveObjects.NotImplementedException;
 
 import java.io.File;
 import java.io.IOException;
@@ -291,13 +289,19 @@ private void initSuggester() {
         }
 
         File suggesterDir = new File(env.getDataRootPath(), IndexDatabase.SUGGESTER_DIR);
+        int rebuildParalleismLevel = (int)(((float)suggesterConfig.getRebuildThreadPoolSizeInNcpuPercent() / 100) * Runtime.getRuntime().availableProcessors());
+        if (rebuildParalleismLevel == 0) {
+            rebuildParalleismLevel = 1;
+        }
+        logger.log(Level.FINER, "Suggester rebuild parallelism level: " + rebuildParalleismLevel);
         suggester = new Suggester(suggesterDir,
                 suggesterConfig.getMaxResults(),
                 Duration.ofSeconds(suggesterConfig.getBuildTerminationTime()),
                 suggesterConfig.isAllowMostPopular(),
                 env.isProjectsEnabled(),
                 suggesterConfig.getAllowedFields(),
-                suggesterConfig.getTimeThreshold());
+                suggesterConfig.getTimeThreshold(),
+                rebuildParalleismLevel);
 
         new Thread(() -> {
             suggester.init(getAllProjectIndexDirs());

suggester/src/main/java/org/opengrok/suggest/Suggester.java

@@ -84,9 +84,16 @@ public final class Suggester implements Closeable {
 
     private final int timeThreshold;
 
+    private final int rebuildParallelismLevel;
+
     // do NOT use fork join thread pool (work stealing thread pool) because it does not send interrupts upon cancellation
     private final ExecutorService executorService = Executors.newFixedThreadPool(
-            Runtime.getRuntime().availableProcessors());
+            Runtime.getRuntime().availableProcessors(),
+            runnable -> {
+                Thread thread = Executors.defaultThreadFactory().newThread(runnable);
+                thread.setName("suggester-lookup-" + thread.getId());
+                return thread;
+            });
 
     /**
      * @param suggesterDir directory under which the suggester data should be created
@@ -105,7 +112,8 @@ public Suggester(
             final boolean allowMostPopular,
             final boolean projectsEnabled,
             final Set<String> allowedFields,
-            final int timeThreshold
+            final int timeThreshold,
+            final int rebuildParallelismLevel
     ) {
         if (suggesterDir == null) {
             throw new IllegalArgumentException("Suggester needs to have directory specified");
@@ -123,25 +131,26 @@ public Suggester(
         this.projectsEnabled = projectsEnabled;
         this.allowedFields = new HashSet<>(allowedFields);
         this.timeThreshold = timeThreshold;
+        this.rebuildParallelismLevel = rebuildParallelismLevel;
     }
 
     /**
      * Initializes suggester data for specified indexes. The data is initialized asynchronously.
-     * @param luceneIndexes paths to lucene indexes and name with which the index should be associated
+     * @param luceneIndexes paths to Lucene indexes and name with which the index should be associated
      */
     public void init(final Collection<NamedIndexDir> luceneIndexes) {
         if (luceneIndexes == null || luceneIndexes.isEmpty()) {
             logger.log(Level.INFO, "No index directories found, exiting...");
             return;
         }
         if (!projectsEnabled && luceneIndexes.size() > 1) {
-            throw new IllegalArgumentException("Projects are not enabled and multiple lucene indexes were passed");
+            throw new IllegalArgumentException("Projects are not enabled and multiple Lucene indexes were passed");
         }
 
         synchronized (lock) {
             logger.log(Level.INFO, "Initializing suggester");
 
-            ExecutorService executor = Executors.newWorkStealingPool();
+            ExecutorService executor = Executors.newWorkStealingPool(rebuildParallelismLevel);
 
             for (NamedIndexDir indexDir : luceneIndexes) {
                 submitInitIfIndexExists(executor, indexDir);
@@ -222,9 +231,9 @@ public void rebuild(final Collection<NamedIndexDir> indexDirs) {
         }
 
         synchronized (lock) {
-            logger.log(Level.INFO, "Rebuilding following suggesters: {0}", indexDirs);
+            logger.log(Level.INFO, "Rebuilding the following suggesters: {0}", indexDirs);
 
-            ExecutorService executor = Executors.newWorkStealingPool();
+            ExecutorService executor = Executors.newWorkStealingPool(rebuildParallelismLevel);
 
             for (NamedIndexDir indexDir : indexDirs) {
                 SuggesterProjectData data = this.projectData.get(indexDir.name);

suggester/src/test/java/org/opengrok/suggest/SuggesterTest.java

@@ -98,14 +98,14 @@ private Suggester.NamedIndexReader getNamedIndexReader() throws IOException {
 
     @Test(expected = IllegalArgumentException.class)
     public void testNullSuggesterDir() {
-        new Suggester(null, 10, Duration.ofMinutes(5), false, true, null, Integer.MAX_VALUE);
+        new Suggester(null, 10, Duration.ofMinutes(5), false, true, null, Integer.MAX_VALUE, 1);
     }
 
     @Test(expected = IllegalArgumentException.class)
     public void testNullDuration() throws IOException {
         Path tempFile = Files.createTempFile("opengrok", "test");
         try {
-            new Suggester(tempFile.toFile(), 10, null, false, true, null, Integer.MAX_VALUE);
+            new Suggester(tempFile.toFile(), 10, null, false, true, null, Integer.MAX_VALUE, 1);
         } finally {
             tempFile.toFile().delete();
         }
@@ -115,7 +115,7 @@ public void testNullDuration() throws IOException {
     public void testNegativeDuration() throws IOException {
         Path tempFile = Files.createTempFile("opengrok", "test");
         try {
-            new Suggester(tempFile.toFile(), 10, Duration.ofMinutes(-4), false, true, null, Integer.MAX_VALUE);
+            new Suggester(tempFile.toFile(), 10, Duration.ofMinutes(-4), false, true, null, Integer.MAX_VALUE, 1);
         } finally {
             tempFile.toFile().delete();
         }
@@ -132,7 +132,7 @@ private SuggesterTestData initSuggester() throws IOException {
         Path tempSuggesterDir = Files.createTempDirectory("opengrok");
 
         Suggester s = new Suggester(tempSuggesterDir.toFile(), 10, Duration.ofMinutes(1), true,
-                true, Collections.singleton("test"), Integer.MAX_VALUE);
+                true, Collections.singleton("test"), Integer.MAX_VALUE, Runtime.getRuntime().availableProcessors());
 
         s.init(Collections.singleton(new Suggester.NamedIndexDir("test", tempIndexDir)));
 
@@ -142,6 +142,7 @@ private SuggesterTestData initSuggester() throws IOException {
         testData.s = s;
         testData.indexDir = tempIndexDir;
         testData.suggesterDir = tempSuggesterDir;
+
         return testData;
     }
 
@@ -204,7 +205,8 @@ public void testIndexChangedWhileOffline() throws IOException {
         addText(t.getIndexDirectory(), "a1 a2");
 
         t.s = new Suggester(t.suggesterDir.toFile(), 10, Duration.ofMinutes(1), false,
-                true, Collections.singleton("test"), Integer.MAX_VALUE);
+                true, Collections.singleton("test"), Integer.MAX_VALUE,
+                Runtime.getRuntime().availableProcessors());
 
         t.s.init(Collections.singleton(t.getNamedIndexDir()));