You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@isis.apache.org by ah...@apache.org on 2019/01/09 13:31:51 UTC
[isis] branch v2 updated: ISIS-2082: let
ThreadPoolSupport.HIGHEST_CONCURRENCY_EXECUTION_MODE_ALLOWED act as an
upper bound for the concurrency level supported by ThreadPoolSupport
This is an automated email from the ASF dual-hosted git repository.
ahuber pushed a commit to branch v2
in repository https://gitbox.apache.org/repos/asf/isis.git
The following commit(s) were added to refs/heads/v2 by this push:
new 9ddf88e ISIS-2082: let ThreadPoolSupport.HIGHEST_CONCURRENCY_EXECUTION_MODE_ALLOWED act as an upper bound for the concurrency level supported by ThreadPoolSupport
9ddf88e is described below
commit 9ddf88e20816743d80bc147ffd54fa3c3d43e444
Author: Andi Huber <ah...@apache.org>
AuthorDate: Wed Jan 9 14:31:41 2019 +0100
ISIS-2082: let
ThreadPoolSupport.HIGHEST_CONCURRENCY_EXECUTION_MODE_ALLOWED act as an
upper bound for the concurrency level supported by ThreadPoolSupport
Task-Url: https://issues.apache.org/jira/browse/ISIS-2082
---
.../metamodel/specloader/SpecificationLoader.java | 10 +-
.../threadpool/ThreadPoolExecutionMode.java | 73 +++++++++++
.../core/runtime/threadpool/ThreadPoolSupport.java | 145 ++++++++++++++-------
.../runtime/threadpool/ThreadPoolSupportTest.java | 15 ++-
.../system/session/IsisSessionFactoryBuilder.java | 4 +-
.../wicket/viewer/IsisWicketApplication.java | 5 +-
6 files changed, 198 insertions(+), 54 deletions(-)
diff --git a/core/metamodel/src/main/java/org/apache/isis/core/metamodel/specloader/SpecificationLoader.java b/core/metamodel/src/main/java/org/apache/isis/core/metamodel/specloader/SpecificationLoader.java
index 4644513..f91dd60 100644
--- a/core/metamodel/src/main/java/org/apache/isis/core/metamodel/specloader/SpecificationLoader.java
+++ b/core/metamodel/src/main/java/org/apache/isis/core/metamodel/specloader/SpecificationLoader.java
@@ -58,6 +58,7 @@ import org.apache.isis.core.metamodel.specloader.specimpl.dflt.ObjectSpecificati
import org.apache.isis.core.metamodel.specloader.specimpl.standalonelist.ObjectSpecificationOnStandaloneList;
import org.apache.isis.core.metamodel.specloader.validator.MetaModelValidator;
import org.apache.isis.core.metamodel.specloader.validator.ValidationFailures;
+import org.apache.isis.core.runtime.threadpool.ThreadPoolExecutionMode;
import org.apache.isis.core.runtime.threadpool.ThreadPoolSupport;
import org.apache.isis.progmodels.dflt.ProgrammingModelFacetsJava5;
import org.apache.isis.schema.utils.CommonDtoUtils;
@@ -272,9 +273,12 @@ public class SpecificationLoader implements ApplicationScopedComponent {
final ThreadPoolSupport threadPoolSupport = ThreadPoolSupport.getInstance();
final boolean parallelize = CONFIG_PROPERTY_PARALLELIZE.from(getConfiguration());
- final List<Future<Object>> futures = parallelize
- ? threadPoolSupport.invokeAll(callables)
- : threadPoolSupport.invokeAllSequential(callables);
+ final ThreadPoolExecutionMode executionModeFromConfig = parallelize
+ ? ThreadPoolExecutionMode.PARALLEL
+ : ThreadPoolExecutionMode.SEQUENTIAL;
+
+ final List<Future<Object>> futures =
+ threadPoolSupport.invokeAll(executionModeFromConfig, callables);
threadPoolSupport.joinGatherFailures(futures);
}
diff --git a/core/metamodel/src/main/java/org/apache/isis/core/runtime/threadpool/ThreadPoolExecutionMode.java b/core/metamodel/src/main/java/org/apache/isis/core/runtime/threadpool/ThreadPoolExecutionMode.java
new file mode 100644
index 0000000..9100cfd
--- /dev/null
+++ b/core/metamodel/src/main/java/org/apache/isis/core/runtime/threadpool/ThreadPoolExecutionMode.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.isis.core.runtime.threadpool;
+
+import org.apache.isis.commons.internal.exceptions._Exceptions;
+
+public enum ThreadPoolExecutionMode {
+
+ /**
+ * Wraps submitted tasks into a single task, which is then executed within the context
+ * of the calling thread. Basically to allow for switching off background execution to
+ * run submitted tasks in sequence (legacy behavior) for debugging purposes.
+ * (expected lowest concurrency)
+ */
+ SEQUENTIAL_WITHIN_CALLING_THREAD,
+
+ /**
+ * Wraps submitted tasks into a single task, which is then executed on the default executor.
+ * (expected medium concurrency)
+ */
+ SEQUENTIAL,
+
+ /**
+ * Executes submitted tasks on the default executor.
+ * (expected highest concurrency)
+ */
+ PARALLEL,
+
+ ;
+
+ /**
+ * {@link ThreadPoolSupport.HIGHEST_CONCURRENCY_EXECUTION_MODE_ALLOWED} acts as an upper bound for the
+ * concurrency level supported.
+ * @param proposedExecutionMode
+ * @return proposedExecutionMode or the highest possible concurrent execution mode allowed, if
+ * the proposedExecutionMode exceeds the highest allowed
+ */
+ public static ThreadPoolExecutionMode honorHighestConcurrencyAllowed(
+ ThreadPoolExecutionMode proposedExecutionMode) {
+
+ switch (ThreadPoolSupport.HIGHEST_CONCURRENCY_EXECUTION_MODE_ALLOWED) {
+ case PARALLEL:
+ return proposedExecutionMode;
+
+ case SEQUENTIAL:
+ return ThreadPoolExecutionMode.SEQUENTIAL;
+
+ case SEQUENTIAL_WITHIN_CALLING_THREAD:
+ return ThreadPoolExecutionMode.SEQUENTIAL_WITHIN_CALLING_THREAD;
+
+ default:
+ throw _Exceptions.unmatchedCase(ThreadPoolSupport.HIGHEST_CONCURRENCY_EXECUTION_MODE_ALLOWED);
+ }
+
+ }
+
+}
diff --git a/core/metamodel/src/main/java/org/apache/isis/core/runtime/threadpool/ThreadPoolSupport.java b/core/metamodel/src/main/java/org/apache/isis/core/runtime/threadpool/ThreadPoolSupport.java
index 7f18d12..67ee062 100644
--- a/core/metamodel/src/main/java/org/apache/isis/core/runtime/threadpool/ThreadPoolSupport.java
+++ b/core/metamodel/src/main/java/org/apache/isis/core/runtime/threadpool/ThreadPoolSupport.java
@@ -19,7 +19,12 @@
package org.apache.isis.core.runtime.threadpool;
-import java.util.Arrays;
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
+import static java.util.stream.Collectors.toList;
+import static org.apache.isis.commons.internal.base._Casts.uncheckedCast;
+import static org.apache.isis.commons.internal.base._With.requires;
+
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -29,6 +34,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
@@ -42,15 +48,9 @@ import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.isis.commons.internal.base._NullSafe;
import org.apache.isis.commons.internal.collections._Lists;
import org.apache.isis.commons.internal.context._Context;
-
-import static java.util.Collections.emptyList;
-import static java.util.Collections.singletonList;
-import static java.util.stream.Collectors.toList;
-import static org.apache.isis.commons.internal.base._Casts.uncheckedCast;
-import static org.apache.isis.commons.internal.base._With.requires;
+import org.apache.isis.commons.internal.exceptions._Exceptions;
/**
* ThreadPoolSupport is application-scoped, meaning ThreadPoolSupport is closed on
@@ -64,6 +64,8 @@ public final class ThreadPoolSupport implements AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(ThreadPoolSupport.class);
+ public static ThreadPoolExecutionMode HIGHEST_CONCURRENCY_EXECUTION_MODE_ALLOWED = ThreadPoolExecutionMode.PARALLEL;
+
private final static int KEEP_ALIVE_TIME_SECS = 5;
private final static int QUEUE_CAPACITY = Integer.MAX_VALUE;
@@ -95,7 +97,6 @@ public final class ThreadPoolSupport implements AutoCloseable {
TimeUnit.SECONDS,
workQueueFactory.get(),
threadFactory);
-
}
/*
@@ -126,58 +127,114 @@ public final class ThreadPoolSupport implements AutoCloseable {
requires(computation, "computation");
return CompletableFuture.supplyAsync(computation, getExecutor());
}
-
/**
* Executes specified {@code callables} on the default executor.
* See {@link ThreadPoolExecutor#invokeAll(java.util.Collection)}
- * @param callables nullable
- * @return non-null
- */
- public List<Future<Object>> invokeAll(@Nullable final List<Callable<Object>> callables) {
- return invokeAll(concurrentExecutor, callables);
- }
-
- /**
- * Executes specified {@code callables} on the default executor.
- * See {@link ThreadPoolExecutor#invokeAll(java.util.Collection)}
- * @param callables nullable
+ * @param proposedExecutionMode - if 'higher concurrency than allowed' is replaced by 'highest-allowed'
+ * @param callables - nullable
* @return non-null
*/
- @SafeVarargs
- public final List<Future<Object>> invokeAll(final Callable<Object>... callables) {
- return invokeAll(Arrays.asList(callables));
- }
+ public List<Future<Object>> invokeAll(
+ final ThreadPoolExecutionMode proposedExecutionMode,
+ @Nullable final List<Callable<Object>> callables) {
- /**
- * Wraps specified {@code callables} into a single task, which is then executed on the default executor.
- * The single task itself executes specified {@code callables} in sequence, one by one.
- * @param callables nullable
- * @return non-null
- */
- public List<Future<Object>> invokeAllSequential(@Nullable final List<Callable<Object>> callables) {
- if(_NullSafe.isEmpty(callables)) {
+ if(isEmpty(callables)) {
return emptyList();
}
- final Future<List<Object>> commonFuture =
- uncheckedCast(invokeAll(singletonList(toSingleTask(callables))).get(0));
- return IntStream.range(0, callables.size())
- .mapToObj(index->new FutureWithIndexIntoFutureOfList<Object>(commonFuture, index))
- .collect(toList());
- }
+ requires(proposedExecutionMode, "proposedExecutionMode");
+
+ final ThreadPoolExecutionMode executionMode =
+ ThreadPoolExecutionMode.honorHighestConcurrencyAllowed(proposedExecutionMode);
+ switch (executionMode) {
+ case PARALLEL:
+ return invokeAll(concurrentExecutor, callables);
+
+ case SEQUENTIAL:
+ {
+ final Future<List<Object>> commonFuture =
+ uncheckedCast(invokeAll(concurrentExecutor,
+ singletonList(toSingleTask(callables))).get(0));
+
+ return IntStream.range(0, callables.size())
+ .mapToObj(index->new FutureWithIndexIntoFutureOfList<Object>(commonFuture, index))
+ .collect(toList());
+ }
+
+ case SEQUENTIAL_WITHIN_CALLING_THREAD:
+ {
+ return callables.stream()
+ .map(FutureTask::new)
+ .peek(FutureTask::run) // immediately run task on submission
+ .collect(toList());
+ }
+
+ default:
+ throw _Exceptions.unmatchedCase(ThreadPoolSupport.HIGHEST_CONCURRENCY_EXECUTION_MODE_ALLOWED);
+ }
+
+ }
+
/**
- * Wraps specified {@code callables} into a single task, which is then executed on the default executor.
- * The single task itself executes specified {@code callables} in sequence, one by one.
+ * Executes specified {@code callables} on the default executor.
+ * See {@link ThreadPoolExecutor#invokeAll(java.util.Collection)}
* @param callables nullable
* @return non-null
*/
- @SafeVarargs
- public final List<Future<Object>> invokeAllSequential(final Callable<Object>... callables) {
- return invokeAllSequential(Arrays.asList(callables));
+ public List<Future<Object>> invokeAll(@Nullable final List<Callable<Object>> callables) {
+ return invokeAll(HIGHEST_CONCURRENCY_EXECUTION_MODE_ALLOWED, callables);
}
+//TODO [ahuber] I'm proposing removal, in order to not bloat this class and since wrapping vargs
+// on the callers side with Arrays.asList(...) is trivial
+// /**
+// * Executes specified {@code callables} on the default executor.
+// * See {@link ThreadPoolExecutor#invokeAll }
+// * @param executionMode
+// * @param callables nullable
+// * @return non-null
+// */
+// @SafeVarargs
+// public final List<Future<Object>> invokeAll(
+// final ThreadPoolExecutionMode executionMode,
+// final Callable<Object>... callables) {
+//
+// requires(executionMode, "executionMode");
+// requires(callables, "callables");
+// return invokeAll(executionMode, Arrays.asList(callables));
+// }
+
+// /**
+// * Wraps specified {@code callables} into a single task, which is then executed on the default executor.
+// * The single task itself executes specified {@code callables} in sequence, one by one.
+// * @param callables nullable
+// * @return non-null
+// */
+// public List<Future<Object>> invokeAllSequential(@Nullable final List<Callable<Object>> callables) {
+// if(_NullSafe.isEmpty(callables)) {
+// return emptyList();
+// }
+// final Future<List<Object>> commonFuture =
+// uncheckedCast(invokeAll(singletonList(toSingleTask(callables))).get(0));
+//
+// return IntStream.range(0, callables.size())
+// .mapToObj(index->new FutureWithIndexIntoFutureOfList<Object>(commonFuture, index))
+// .collect(toList());
+// }
+//
+// /**
+// * Wraps specified {@code callables} into a single task, which is then executed on the default executor.
+// * The single task itself executes specified {@code callables} in sequence, one by one.
+// * @param callables nullable
+// * @return non-null
+// */
+// @SafeVarargs
+// public final List<Future<Object>> invokeAllSequential(final Callable<Object>... callables) {
+// return invokeAllSequential(Arrays.asList(callables));
+// }
+
/**
* Waits if necessary for the computation to complete. Suppresses checked exceptions.
diff --git a/core/metamodel/src/test/java/org/apache/isis/core/runtime/threadpool/ThreadPoolSupportTest.java b/core/metamodel/src/test/java/org/apache/isis/core/runtime/threadpool/ThreadPoolSupportTest.java
index ae6372f..3d7b617 100644
--- a/core/metamodel/src/test/java/org/apache/isis/core/runtime/threadpool/ThreadPoolSupportTest.java
+++ b/core/metamodel/src/test/java/org/apache/isis/core/runtime/threadpool/ThreadPoolSupportTest.java
@@ -58,6 +58,17 @@ class ThreadPoolSupportTest {
@Test
void shouldPreserveSequence_whenSequentialExecution() {
+ runSequentialExecutionWithMode(ThreadPoolExecutionMode.SEQUENTIAL);
+ }
+
+ @Test
+ void shouldPreserveSequence_whenSequentialExecutionWithinCallingThread() {
+ runSequentialExecutionWithMode(ThreadPoolExecutionMode.SEQUENTIAL_WITHIN_CALLING_THREAD);
+ }
+
+ // -- HELPER
+
+ void runSequentialExecutionWithMode(ThreadPoolExecutionMode executionMode) {
final List<Future<Object>> futures = _Lists.newArrayList();
@@ -75,7 +86,7 @@ class ThreadPoolSupportTest {
);
for(int i=0; i<256; ++i) {
- final List<Future<Object>> taskFutures = pool.invokeAllSequential(tasks);
+ final List<Future<Object>> taskFutures = pool.invokeAll(executionMode, tasks);
pool.joinGatherFailures(taskFutures);
futures.addAll(taskFutures);
}
@@ -98,8 +109,6 @@ class ThreadPoolSupportTest {
}
- // -- HELPER
-
/** thread-safe StringBuffer */
private static class Buffer {
diff --git a/core/runtime/src/main/java/org/apache/isis/core/runtime/system/session/IsisSessionFactoryBuilder.java b/core/runtime/src/main/java/org/apache/isis/core/runtime/system/session/IsisSessionFactoryBuilder.java
index f063137..014347b 100644
--- a/core/runtime/src/main/java/org/apache/isis/core/runtime/system/session/IsisSessionFactoryBuilder.java
+++ b/core/runtime/src/main/java/org/apache/isis/core/runtime/system/session/IsisSessionFactoryBuilder.java
@@ -167,7 +167,7 @@ public class IsisSessionFactoryBuilder {
_Context.putSingleton(IsisSessionFactory.class, isisSessionFactory);
// execute tasks using a threadpool
- final List<Future<Object>> futures = ThreadPoolSupport.getInstance().invokeAll(
+ final List<Future<Object>> futures = ThreadPoolSupport.getInstance().invokeAll(Arrays.asList(
new Callable<Object>() {
@Override
public Object call() {
@@ -238,7 +238,7 @@ public class IsisSessionFactoryBuilder {
return "CommandDtoUtils.init()";
}
}
- );
+ ));
// wait on this thread for tasks to complete
diff --git a/core/viewer-wicket-impl/src/main/java/org/apache/isis/viewer/wicket/viewer/IsisWicketApplication.java b/core/viewer-wicket-impl/src/main/java/org/apache/isis/viewer/wicket/viewer/IsisWicketApplication.java
index 955174d..0c602bf 100644
--- a/core/viewer-wicket-impl/src/main/java/org/apache/isis/viewer/wicket/viewer/IsisWicketApplication.java
+++ b/core/viewer-wicket-impl/src/main/java/org/apache/isis/viewer/wicket/viewer/IsisWicketApplication.java
@@ -21,6 +21,7 @@ package org.apache.isis.viewer.wicket.viewer;
import java.io.IOException;
import java.net.URL;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.ServiceLoader;
@@ -371,11 +372,11 @@ implements ComponentFactoryRegistryAccessor, PageClassRegistryAccessor, WicketVi
}
protected List<Future<Object>> startBackgroundInitializationThreads() {
- return ThreadPoolSupport.getInstance().invokeAll(
+ return ThreadPoolSupport.getInstance().invokeAll(Arrays.asList(
Executors.callable(this::configureWebJars),
Executors.callable(this::configureWicketBootstrap),
Executors.callable(this::configureWicketSelect2)
- );
+ ));
}
/**