001 /*
002 * Java Genetic Algorithm Library (jenetics-3.5.0).
003 * Copyright (c) 2007-2016 Franz Wilhelmstötter
004 *
005 * Licensed under the Apache License, Version 2.0 (the "License");
006 * you may not use this file except in compliance with the License.
007 * You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 *
017 * Author:
018 * Franz Wilhelmstötter (franz.wilhelmstoetter@gmx.at)
019 */
020 package org.jenetics.internal.util;
021
022 import static java.util.Objects.requireNonNull;
023
024 import java.util.List;
025 import java.util.concurrent.CancellationException;
026 import java.util.concurrent.ExecutionException;
027 import java.util.concurrent.Executor;
028 import java.util.concurrent.ExecutorService;
029 import java.util.concurrent.ForkJoinPool;
030 import java.util.concurrent.ForkJoinTask;
031 import java.util.concurrent.Future;
032 import java.util.concurrent.FutureTask;
033
034 import org.jenetics.internal.collection.Stack;
035
036 /**
037 * @author <a href="mailto:franz.wilhelmstoetter@gmx.at">Franz Wilhelmstötter</a>
038 * @version 2.0
039 * @since 2.0
040 */
041 public abstract class Concurrency implements Executor, AutoCloseable {
042
043 public static final int CORES = Runtime.getRuntime().availableProcessors();
044
045 public static final Concurrency SERIAL_EXECUTOR = new SerialConcurrency();
046
047 public abstract void execute(final List<? extends Runnable> runnables);
048
049 @Override
050 public abstract void close();
051
052 /**
053 * Return the underlying {@code Executor}, which is used for performing the
054 * actual task execution.
055 *
056 * @return the underlying {@code Executor} object
057 */
058 public Executor getInnerExecutor() {
059 return this;
060 }
061
062 /**
063 * Return an new Concurrency object from the given executor.
064 *
065 * @param executor the underlying Executor
066 * @return a new Concurrency object
067 */
068 public static Concurrency with(final Executor executor) {
069 if (executor instanceof ForkJoinPool) {
070 return new ForkJoinPoolConcurrency((ForkJoinPool)executor);
071 } else if (executor instanceof ExecutorService) {
072 return new ExecutorServiceConcurrency((ExecutorService)executor);
073 } else if (executor == SERIAL_EXECUTOR) {
074 return SERIAL_EXECUTOR;
075 } else {
076 return new ExecutorConcurrency(executor);
077 }
078 }
079
080 /**
081 * Return a new Concurrency object using the common ForkJoinPool.
082 *
083 * @return a new Concurrency object using the new ForkJoinPool
084 */
085 public static Concurrency withCommonPool() {
086 return with(ForkJoinPool.commonPool());
087 }
088
089
090 /**
091 * This Concurrency uses a ForkJoinPool.
092 */
093 private static final class ForkJoinPoolConcurrency extends Concurrency {
094 private final Stack<ForkJoinTask<?>> _tasks = new Stack<>();
095 private final ForkJoinPool _pool;
096
097 ForkJoinPoolConcurrency(final ForkJoinPool pool) {
098 _pool = requireNonNull(pool);
099 }
100
101 @Override
102 public void execute(final Runnable runnable) {
103 _tasks.push(_pool.submit(runnable));
104 }
105
106 @Override
107 public void execute(final List<? extends Runnable> runnables) {
108 _tasks.push(_pool.submit(new RunnablesAction(runnables)));
109 }
110
111 @Override
112 public Executor getInnerExecutor() {
113 return _pool;
114 }
115
116 @Override
117 public void close() {
118 for (ForkJoinTask<?> t = _tasks.pop(); t != null; t = _tasks.pop()) {
119 t.join();
120 }
121 }
122 }
123
124 /**
125 * This Concurrency uses an ExecutorService.
126 */
127 private static final class ExecutorServiceConcurrency extends Concurrency {
128 private final Stack<Future<?>> _futures = new Stack<>();
129 private final ExecutorService _service;
130
131 ExecutorServiceConcurrency(final ExecutorService service) {
132 _service = requireNonNull(service);
133 }
134
135 @Override
136 public void execute(final Runnable command) {
137 _futures.push(_service.submit(command));
138 }
139
140 @Override
141 public void execute(final List<? extends Runnable> runnables) {
142 final int[] parts = partition(runnables.size(), CORES + 1);
143 for (int i = 0; i < parts.length - 1; ++i) {
144 execute(new RunnablesRunnable(runnables, parts[i], parts[i + 1]));
145 }
146 }
147
148 @Override
149 public Executor getInnerExecutor() {
150 return _service;
151 }
152
153 @Override
154 public void close() {
155 try {
156 for (Future<?> f = _futures.pop(); f != null; f = _futures.pop()) {
157 f.get();
158 }
159 } catch (InterruptedException|ExecutionException e) {
160 final String msg = e.getMessage();
161 throw (CancellationException)new CancellationException(msg)
162 .initCause(e);
163 }
164 }
165 }
166
167 /**
168 * This Concurrency uses an Executor.
169 */
170 private static final class ExecutorConcurrency extends Concurrency {
171 private final Stack<FutureTask<?>> _tasks = new Stack<>();
172 private final Executor _executor;
173
174 ExecutorConcurrency(final Executor executor) {
175 _executor = requireNonNull(executor);
176 }
177
178 @Override
179 public void execute(final Runnable command) {
180 final FutureTask<?> task = new FutureTask<>(command, null);
181 _tasks.push(task);
182 _executor.execute(task);
183 }
184
185 @Override
186 public void execute(final List<? extends Runnable> runnables) {
187 final int[] parts = partition(runnables.size(), CORES + 1);
188 for (int i = 0; i < parts.length - 1; ++i) {
189 execute(new RunnablesRunnable(runnables, parts[i], parts[i + 1]));
190 }
191 }
192
193 @Override
194 public void close() {
195 try {
196 for (FutureTask<?> t = _tasks.pop(); t != null; t = _tasks.pop()) {
197 t.get();
198 }
199 } catch (InterruptedException|ExecutionException e) {
200 final String msg = e.getMessage();
201 throw (CancellationException)new CancellationException(msg)
202 .initCause(e);
203 }
204 }
205 }
206
207 /**
208 * This Concurrency executes the runnables within the main thread.
209 */
210 private static final class SerialConcurrency extends Concurrency {
211
212 @Override
213 public void execute(final Runnable command) {
214 command.run();
215 }
216
217 @Override
218 public void execute(final List<? extends Runnable> runnables) {
219 for (final Runnable runnable : runnables) {
220 runnable.run();
221 }
222 }
223
224 @Override
225 public void close() {
226 }
227 }
228
229
230 /**
231 * Return a array with the indexes of the partitions of an array with the
232 * given size. The length of the returned array is {@code min(size, prts) + 1}.
233 * <p>
234 * Some examples:
235 * <pre>
236 * partition(10, 3): [0, 3, 6, 10]
237 * partition(15, 6): [0, 2, 4, 6, 9, 12, 15]
238 * partition(5, 10): [0, 1, 2, 3, 4, 5]
239 * </pre>
240 *
241 * The following examples prints the start index (inclusive) and the end
242 * index (exclusive) of the {@code partition(15, 6)}.
243 * <pre>{@code
244 * int[] parts = partition(15, 6);
245 * for (int i = 0; i < parts.length - 1; ++i) {
246 * System.out.println(i + ": " + parts[i] + "\t" + parts[i + 1]);
247 * }
248 * }</pre>
249 * <pre>
250 * 0: 0 2
251 * 1: 2 4
252 * 2: 4 6
253 * 3: 6 9
254 * 4: 9 12
255 * 5: 12 15
256 * </pre>
257 *
258 * This example shows how this can be used in an concurrent environment:
259 * <pre>{@code
260 * try (final Concurrency c = Concurrency.start()) {
261 * final int[] parts = arrays.partition(population.size(), _maxThreads);
262 *
263 * for (int i = 0; i < parts.length - 1; ++i) {
264 * final int part = i;
265 * c.execute(new Runnable() { @Override public void run() {
266 * for (int j = parts[part + 1]; --j <= parts[part];) {
267 * population.get(j).evaluate();
268 * }
269 * }});
270 * }
271 * }
272 * }</pre>
273 *
274 * @param size the size of the array to partition.
275 * @param parts the number of parts the (virtual) array should be partitioned.
276 * @return the partition array with the length of {@code min(size, parts) + 1}.
277 * @throws IllegalArgumentException if {@code size} or {@code p} is less than one.
278 */
279 private static int[] partition(final int size, final int parts) {
280 if (size < 1) {
281 throw new IllegalArgumentException(
282 "Size must greater than zero: " + size
283 );
284 }
285 if (parts < 1) {
286 throw new IllegalArgumentException(
287 "Number of partitions must greater than zero: " + parts
288 );
289 }
290
291 final int pts = Math.min(size, parts);
292 final int[] partition = new int[pts + 1];
293
294 final int bulk = size/pts;
295 final int rest = size%pts;
296 assert (bulk*pts + rest) == size;
297
298 for (int i = 0, n = pts - rest; i < n; ++i) {
299 partition[i] = i*bulk;
300 }
301 for (int i = 0, n = rest + 1; i < n; ++i) {
302 partition[pts - rest + i] = (pts - rest)*bulk + i*(bulk + 1);
303 }
304
305 return partition;
306 }
307
308 }
|