diff --git a/community/consistency-check/src/main/java/org/neo4j/consistency/internal/SchemaIndexExtensionLoader.java b/community/consistency-check/src/main/java/org/neo4j/consistency/internal/SchemaIndexExtensionLoader.java index 14134b4e97f2d..2a240c1d345a5 100644 --- a/community/consistency-check/src/main/java/org/neo4j/consistency/internal/SchemaIndexExtensionLoader.java +++ b/community/consistency-check/src/main/java/org/neo4j/consistency/internal/SchemaIndexExtensionLoader.java @@ -41,8 +41,16 @@ import org.neo4j.kernel.impl.util.Dependencies; import org.neo4j.kernel.lifecycle.LifecycleAdapter; +/** + * Utility for loading {@link SchemaIndexProvider} instances from {@link KernelExtensions}. + */ public class SchemaIndexExtensionLoader { + /** + * Used in scenarios where recovery isn't allowed. + */ + public static final RecoveryCleanupWorkCollector RECOVERY_PREVENTING_COLLECTOR = new RecoveryPreventingCollector(); + public static SchemaIndexProviderMap loadSchemaIndexProviders( KernelExtensions extensions ) { AllByPrioritySelectionStrategy indexProviderSelection = new AllByPrioritySelectionStrategy<>(); @@ -76,6 +84,4 @@ public void add( CleanupJob job ) } } } - - public static final RecoveryCleanupWorkCollector RECOVERY_PREVENTING_COLLECTOR = new RecoveryPreventingCollector(); } diff --git a/community/kernel/src/main/java/org/neo4j/kernel/extension/dependency/AllByPrioritySelectionStrategy.java b/community/kernel/src/main/java/org/neo4j/kernel/extension/dependency/AllByPrioritySelectionStrategy.java index c2d124682aa4e..42dc2694aca04 100644 --- a/community/kernel/src/main/java/org/neo4j/kernel/extension/dependency/AllByPrioritySelectionStrategy.java +++ b/community/kernel/src/main/java/org/neo4j/kernel/extension/dependency/AllByPrioritySelectionStrategy.java @@ -30,17 +30,20 @@ /** * Selects the candidate with highest priority (assumed to implement {@link Comparable}) and returns * in {@link #select(Class, Iterable)}, but keeps the others for access too. + * + * @param type of items expected to be provided into {@link #select(Class, Iterable)}. Due to signature of the + * {@link #select(Class, Iterable) select method} where an explicit and local {@code R} is defined a cast from + * {@code R} to {@code T} is required and so will fail if {@code T} isn't matching {@code R}. */ -public class AllByPrioritySelectionStrategy implements DependencyResolver.SelectionStrategy +public class AllByPrioritySelectionStrategy> implements DependencyResolver.SelectionStrategy { - @SuppressWarnings( "rawtypes" ) - private List lowerPrioritizedCandidates = Collections.emptyList(); + private List lowerPrioritizedCandidates = Collections.emptyList(); - @SuppressWarnings( {"rawtypes", "unchecked"} ) + @SuppressWarnings( "unchecked" ) @Override public R select( Class type, Iterable candidates ) throws IllegalArgumentException { - List all = (List) Iterables.asList( candidates ); + List all = (List) Iterables.asList( candidates ); if ( all.isEmpty() ) { throw new IllegalArgumentException( "Could not resolve dependency of type: " + @@ -52,7 +55,6 @@ public R select( Class type, Iterable candidates ) throws IllegalArgum return highest; } - @SuppressWarnings( "unchecked" ) public Iterable lowerPrioritizedCandidates() { return lowerPrioritizedCandidates; diff --git a/community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/BatchingMultipleIndexPopulator.java b/community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/BatchingMultipleIndexPopulator.java index 992a822af993b..b297f56ab539a 100644 --- a/community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/BatchingMultipleIndexPopulator.java +++ b/community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/BatchingMultipleIndexPopulator.java @@ -64,6 +64,8 @@ public class BatchingMultipleIndexPopulator extends MultipleIndexPopulator private static final String EOL = System.lineSeparator(); private static final String FLUSH_THREAD_NAME_PREFIX = "Index Population Flush Thread"; + static final int BATCH_SIZE = FeatureToggles.getInteger( BatchingMultipleIndexPopulator.class, BATCH_SIZE_NAME, 10_000 ); + private final int MAXIMUM_NUMBER_OF_WORKERS = FeatureToggles.getInteger( getClass(), MAXIMUM_NUMBER_OF_WORKERS_NAME, Runtime.getRuntime().availableProcessors() - 1 ); private final int TASK_QUEUE_SIZE = FeatureToggles.getInteger( getClass(), TASK_QUEUE_SIZE_NAME, @@ -151,39 +153,23 @@ private void awaitCompletion() } } - /** - * Add given {@link IndexEntryUpdate update} to the list of updates already present for the given - * {@link IndexPopulation population}. Flushes all updates if {@link #BATCH_SIZE} is reached. - * - * @param population the index population. - * @param update updates to add to the batch. - */ - private void batchUpdate( IndexPopulation population, IndexEntryUpdate update ) - { - if ( population.batch( update ) ) - { - Collection> batchedUpdates = population.takeCurrentBatch(); - flush( population, batchedUpdates ); - } - } - /** * Insert all batched updates into corresponding indexes. */ private void flushAll() { - populations.forEach( population -> flush( population, population.takeCurrentBatch() ) ); + populations.forEach( population -> flush( population ) ); } /** * Insert the given batch of updates into the index defined by the given {@link IndexPopulation}. * * @param population the index population. - * @param batch the list of updates to insert. */ - private void flush( IndexPopulation population, Collection> batch ) + private void flush( IndexPopulation population ) { activeTasks.incrementAndGet(); + Collection> batch = population.takeCurrentBatch(); executor.execute( () -> { diff --git a/community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/MultipleIndexPopulator.java b/community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/MultipleIndexPopulator.java index 6458bfa9d8b88..a60bed575ef68 100644 --- a/community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/MultipleIndexPopulator.java +++ b/community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/MultipleIndexPopulator.java @@ -55,7 +55,7 @@ import static java.lang.String.format; import static org.neo4j.collection.primitive.PrimitiveIntCollections.contains; -import static org.neo4j.kernel.impl.api.index.BatchingMultipleIndexPopulator.BATCH_SIZE_NAME; +import static org.neo4j.kernel.impl.api.index.BatchingMultipleIndexPopulator.BATCH_SIZE; import static org.neo4j.kernel.impl.api.index.IndexPopulationFailure.failure; /** @@ -86,10 +86,8 @@ */ public class MultipleIndexPopulator implements IndexPopulator { - public static final String QUEUE_THRESHOLD_NAME = "queue_threshold"; private final int QUEUE_THRESHOLD = FeatureToggles.getInteger( getClass(), QUEUE_THRESHOLD_NAME, 20_000 ); - private final int BATCH_SIZE = FeatureToggles.getInteger( BatchingMultipleIndexPopulator.class, BATCH_SIZE_NAME, 10_000 ); // Concurrency queue since multiple concurrent threads may enqueue updates into it. It is important for this queue // to have fast #size() method since it might be drained in batches