eclipse-sailing-analytics/sailing-analytics
1
0
Fork 0
mirror of https://github.com/eclipse-sailing-analytics/sailing-analytics.git synced 2026-09-20 04:35:32 +00:00
Code Activity

Merge branch 'master' into bug4339

Browse Source
This commit is contained in:
Steffen Schaefer
2018-07-31 08:29:17 +02:00
parent 328d05b4a0 fd1f79e802
commit f02f1e90d5
24 changed files with 788 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestAbortingHeavyLoadQuery.java
+508
View File
@@ -0,0 +1,508 @@
package com.sap.sse.datamining.impl;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.lang.reflect.Method;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.Query;
import com.sap.sse.datamining.QueryState;
import com.sap.sse.datamining.components.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.components.ProcessorInstruction;
import com.sap.sse.datamining.components.ProcessorInstructionHandler;
import com.sap.sse.datamining.data.QueryResult;
import com.sap.sse.datamining.impl.components.AbstractParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
import com.sap.sse.datamining.impl.components.GroupedDataEntry;
import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDataCollectingAsSetProcessor;
import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.data.QueryResultState;
import com.sap.sse.datamining.shared.impl.GenericGroupKey;
import com.sap.sse.datamining.test.util.components.StatefulBlockingInstruction;
import com.sap.sse.datamining.test.util.components.StatefulProcessorInstruction;
/**
* Integration test aborting a query with simulated heavy load instructions. Uses a highly customized processor chain
* (based on the processors/instructions used in production) that allows detailed introspection and logging of the
* execution process. The test is used to verify the behavior of processors, if such a heavy load query is aborted
* during the process. For example that no instructions are scheduled after the query was aborted and the behavior of
* already scheduled instructions that may or may not have been started.<br>
* <br>
* The test is configurable regarding the number of threads for the executor, the number of data elements/heavy load
* instructions, the duration of heavy load instructions, the time to wait before the query is aborted and the time
* given to the executor to finish the remaining instructions. Note that the test can fail for some configurations, due
* to race conditions that would be very hard to check for.<br>
* <br>
* The test optionally records the process in order of the execution (using a concurrent message queue), which is
* printed to the console before assertions are performed. This isn't used by the assertions, but can be used to get a
* better understanding of what happens during the execution.<br>
* <br>
* See {@link #initialize()} for more details about the processor chain.
*
* @author Lennart Hensler
*/
public class TestAbortingHeavyLoadQuery {
// Test Configuration ----------------------------------------------------------------------------------------
/**
* Number of threads in the executor. Determines the number of elements (and thus heavy load instructions)
* retrieved for each group. This means that each group should have a runtime of {@value #HeavyLoadInstructionDuration}ms,
* since the single instructions are executed concurrently.
*/
private static final int ExecutorPoolSize = Math.max(3, Runtime.getRuntime().availableProcessors());
/** The number of groups contained in the initial data source. */
private static final int DataSourceSize = 2000;
private static final String GroupKeyPrefix = "G";
/** The time a heavy load instruction blocks the executing thread using {@link Thread#sleep(long)}. */
private static final long HeavyLoadInstructionDuration = 500;
/** The number of milliseconds to wait before {@link Query#abort()} is called. */
private static final long AbortQueryDelay = (long) (HeavyLoadInstructionDuration * 5.45);
/** The time given to the executor to complete all unfinished instructions */
private static final long TerminationTimeout = (long) (HeavyLoadInstructionDuration * 1.5);
//------------------------------------------------------------------------------------------------------------
// Execution Recording Configuration -------------------------------------------------------------------------
/** Enables concurrent logging of the query execution and prints the current record before assertions. */
private static final boolean RecordExecution = false;
private static final SimpleDateFormat DateFormatter = new SimpleDateFormat("HH:mm:ss.SSS");
private ConcurrentLinkedQueue<String> executionRecord;
//------------------------------------------------------------------------------------------------------------
// Test State - Initialized in initialize() ------------------------------------------------------------------
private ExecutorService executor;
private Query<HashSet<Integer>> query;
private Collection<Processor<?, ?>> processors;
private Set<StatefulProcessorInstruction<?>> unfinishedInstructions;
//------------------------------------------------------------------------------------------------------------
@Test
public void testAbortingHeavyLoadQuery() throws InterruptedException, ExecutionException {
// Executing query in separate thread
RunnableFuture<QueryResult<HashSet<Integer>>> queryTask = new FutureTask<>(() -> {
logExecution("Starting query execution");
long start = System.currentTimeMillis();
QueryResult<HashSet<Integer>> result = query.run();
long duration = System.currentTimeMillis() - start;
logExecution("Finished query in " + duration + "ms");
return result;
});
Thread worker = new Thread(queryTask, "Worker");
worker.start();
do {
Thread.sleep(10);
} while (query.getState() != QueryState.RUNNING);
// Aborting query and waiting for completion
Thread.sleep(AbortQueryDelay);
logExecution("Aborting query");
query.abort();
QueryResult<HashSet<Integer>> result = queryTask.get();
// The query execution returned, so all processors are aborted (checked below).
// The number of unfinished instructions mustn't change after this point
int unfinishedInstructionsCount = unfinishedInstructions.size();
boolean unfinishedInstructionWasFinished = false;
Set<StatefulProcessorInstruction<?>> runningInstructions = new HashSet<>();
Set<StatefulProcessorInstruction<?>> notStartedInstructions = new HashSet<>();
for (StatefulProcessorInstruction<?> instruction : unfinishedInstructions) {
if (instruction.runWasCalled()) {
runningInstructions.add(instruction);
} else {
notStartedInstructions.add(instruction);
}
unfinishedInstructionWasFinished |= instruction.computeResultWasFinished();
}
logExecution(unfinishedInstructionsCount + " unfinished instructions left - " +
runningInstructions.size() + " running, " + notStartedInstructions.size() + " not started");
printExecutionRecord();
assertThat("Number of unfinished instructions changed", runningInstructions.size() + notStartedInstructions.size(), is(unfinishedInstructionsCount));
assertFalse("Unfinished instructions expected, but at least one was completed", unfinishedInstructionWasFinished);
// Checking query, result and processor chain state
assertThat(query.getState(), is(QueryState.ABORTED));
assertThat(result.getState(), is(QueryResultState.ABORTED));
assertTrue("The result is not empty", result.isEmpty());
for (Processor<?, ?> processor : processors) {
assertTrue("Processor wasn't aborted", processor.isAborted());
}
// Execution of unfinished instructions
executor.shutdown();
boolean terminated = executor.awaitTermination(TerminationTimeout, TimeUnit.MILLISECONDS);
printExecutionRecord();
assertTrue("The executor didn't terminate in the given time", terminated);
// Verify that the collection of unfinished instructions didn't change
assertThat("Number of unfinished instructions changed", unfinishedInstructions.size(), is(unfinishedInstructionsCount));
for (StatefulProcessorInstruction<?> instruction : unfinishedInstructions) {
boolean previouslyContained = runningInstructions.contains(instruction) || notStartedInstructions.contains(instruction);
assertTrue("A new instruction has been scheduled after aborting the query", previouslyContained);
}
for (StatefulProcessorInstruction<?> instruction : runningInstructions) {
assertTrue("A previously running instruction was removed from unfinished instructions", unfinishedInstructions.contains(instruction));
}
for (StatefulProcessorInstruction<?> instruction : notStartedInstructions) {
assertTrue("A previously unstarted instruction was removed from unfinished instructions", unfinishedInstructions.contains(instruction));
}
// Checking state of unfinished instructions
for (StatefulProcessorInstruction<?> instruction : runningInstructions) {
assertTrue("computeResult() of a running unfinished instruction wasn't called", instruction.computeResultWasCalled());
assertTrue("computeResult() of a running unfinished instruction didn't finish", instruction.computeResultWasFinished());
}
for (StatefulProcessorInstruction<?> instruction : notStartedInstructions) {
assertTrue("run() of an unstarted unfinished instruction wasn't called", instruction.runWasCalled());
assertFalse("computeResult() of an unstarted unfinished instruction was called", instruction.computeResultWasCalled());
assertFalse("computeResult() of an unstarted unfinished instruction was finished", instruction.computeResultWasFinished());
}
}
private void printExecutionRecord() {
if (RecordExecution) {
Collection<String> snapshot = new ArrayList<>(executionRecord);
executionRecord.clear();
for (String string : snapshot) {
System.out.println(string);
}
}
}
private void logExecution(String message) {
if (RecordExecution) {
String timeString = DateFormatter.format(new Date());
executionRecord.add(timeString + " - " + Thread.currentThread().getName() + ": " + message);
}
}
/**
* Initializes the {@link ExecutorService}, the execution record, helper collections (e.g. the processors in the
* chain or the unfinished instructions) and most importantly the processor chain for the query, which works as
* follows:
* <ol>
* <li>
* The initial data source is a collection of strings from {@value #GroupKeyPrefix}<code>n</code> to
* {@value #GroupKeyPrefix}<code>n-1</code>, denoting a group. <code>n</code> is specified in {@link #DataSourceSize}.
* </li>
* <li>
* <code>x</code> (specified by {@link #ExecutorPoolSize}) {@link Element elements} are retrieved for each
* group, where the elements name is set to the received string and the elements value is the current value
* of <code>x</code>.
* </li>
* <li>
* A heavy load instruction for each element is scheduled, which blocks the running thread for
* {@value #HeavyLoadInstructionDuration}ms.
* </li>
* <li>
* Each element is grouped by its name and its value is used as value for the {@link GroupedDataEntry}.
* </li>
* <li>The grouped data is collected as set.</li>
* </ol>
* This results in <code>n * x</code> data elements (and heavy load instructions), with <code>x</code> instructions
* executed concurrently.<br>
* <br>
* Each processor creates {@link StatefulProcessorInstruction} or uses {@link StatefulInstructionWrapper}, which allows
* to verify if <code>instruction.run()</code> was called, its computation started and its computation finished.<br>
* <br>
* A concurrent set ({@link ConcurrentHashMap#newKeySet()}) is used to track the unfinished instructions. An instruction
* is added to the set upon its construction and is removed when the instruction finished callback method is called.
*/
@Before
@SuppressWarnings("unchecked")
public void initialize() {
if (RecordExecution) {
executionRecord = new ConcurrentLinkedQueue<>();
}
executor = new DataMiningExecutorService(ExecutorPoolSize);
// executor = new ThreadPoolExecutor(ExecutorPoolSize, ExecutorPoolSize, 0, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<>());
// executor = new ThreadPoolExecutor(ExecutorPoolSize, ExecutorPoolSize, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
processors = new ArrayList<>();
unfinishedInstructions = ConcurrentHashMap.newKeySet();
Class<Iterable<String>> dataSourceType = (Class<Iterable<String>>)(Class<?>) Iterable.class;
Class<GroupedDataEntry<Integer>> groupedType = (Class<GroupedDataEntry<Integer>>)(Class<?>) GroupedDataEntry.class;
Class<HashSet<Integer>> resultType = (Class<HashSet<Integer>>)(Class<?>) HashSet.class;
Collection<String> dataSource = new ArrayList<>(DataSourceSize);
for (int i = 0; i < DataSourceSize; i++) {
dataSource.add(GroupKeyPrefix + i);
}
query = new ProcessorQuery<HashSet<Integer>, Iterable<String>>(dataSource, resultType) {
@Override
protected Processor<Iterable<String>, ?> createChainAndReturnFirstProcessor(Processor<Map<GroupKey, HashSet<Integer>>, Void> resultReceiver) {
Processor<GroupedDataEntry<Integer>, Map<GroupKey, HashSet<Integer>>> aggregator = new ParallelGroupedDataCollectingAsSetProcessor<Integer>(executor, Collections.singleton(resultReceiver)) {
@Override
protected ProcessorInstruction<Map<GroupKey, HashSet<Integer>>> createInstruction(GroupedDataEntry<Integer> element) {
AbstractProcessorInstruction<Map<GroupKey, HashSet<Integer>>> instruction = (AbstractProcessorInstruction<Map<GroupKey, HashSet<Integer>>>) super.createInstruction(element);
StatefulProcessorInstruction<Map<GroupKey, HashSet<Integer>>> statefulInstruction = new StatefulInstructionWrapper<>(instruction);
unfinishedInstructions.add(statefulInstruction);
return statefulInstruction;
}
@Override
protected void storeElement(GroupedDataEntry<Integer> element) {
logExecution("Storing " + element);
super.storeElement(element);
}
@Override
public void afterInstructionFinished(ProcessorInstruction<Map<GroupKey, HashSet<Integer>>> instruction) {
super.afterInstructionFinished(instruction);
if(canProcessElements()) unfinishedInstructions.remove(instruction);
}
};
Processor<Element, GroupedDataEntry<Integer>> grouper = new AbstractParallelProcessor<Element, GroupedDataEntry<Integer>>(Element.class, groupedType, executor, Collections.singleton(aggregator)) {
@Override
protected ProcessorInstruction<GroupedDataEntry<Integer>> createInstruction(Element element) {
StatefulProcessorInstruction<GroupedDataEntry<Integer>> instruction = new StatefulProcessorInstruction<GroupedDataEntry<Integer>>(this, ProcessorInstructionPriority.Grouping) {
@Override
protected GroupedDataEntry<Integer> internalComputeResult() throws Exception {
logExecution("Grouping " + element);
return new GroupedDataEntry<>(new GenericGroupKey<>(element.getName()), element.getValue());
}
};
unfinishedInstructions.add(instruction);
return instruction;
}
@Override
public void afterInstructionFinished(ProcessorInstruction<GroupedDataEntry<Integer>> instruction) {
super.afterInstructionFinished(instruction);
if(canProcessElements()) unfinishedInstructions.remove(instruction);
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { }
};
Processor<Element, Element> heavyLoadProcessor = new AbstractParallelProcessor<Element, Element>(Element.class, Element.class, executor, Collections.singleton(grouper)) {
@Override
protected ProcessorInstruction<Element> createInstruction(Element element) {
StatefulProcessorInstruction<Element> instruction = new HeavyLoadInstruction(this,
ProcessorInstructionPriority.Extraction, HeavyLoadInstructionDuration, element,
TestAbortingHeavyLoadQuery.this::logExecution);
unfinishedInstructions.add(instruction);
return instruction;
}
@Override
public void afterInstructionFinished(ProcessorInstruction<Element> instruction) {
super.afterInstructionFinished(instruction);
if(canProcessElements()) unfinishedInstructions.remove(instruction);
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { }
};
Processor<String, Element> retriever1 = new AbstractRetrievalProcessor<String, Element>(String.class, Element.class, executor, Collections.singleton(heavyLoadProcessor), 1) {
@Override
protected ProcessorInstruction<Element> createInstruction(String element) {
AbstractProcessorInstruction<Element> instruction = (AbstractProcessorInstruction<Element>) super.createInstruction(element);
StatefulProcessorInstruction<Element> statefulInstruction = new StatefulInstructionWrapper<>(instruction);
unfinishedInstructions.add(statefulInstruction);
return statefulInstruction;
}
@Override
protected Iterable<Element> retrieveData(String element) {
int count = ExecutorPoolSize;
logExecution("Retrieving " + count + " elements for " + element);
Collection<Element> data = new ArrayList<>();
for (int i = 0; i < count; i++) {
data.add(new Element(element, i));
}
return data;
}
@Override
public void instructionSucceeded(Element result) {
logExecution("Group retrieval finished");
super.instructionSucceeded(result);
}
@Override
public void afterInstructionFinished(ProcessorInstruction<Element> instruction) {
super.afterInstructionFinished(instruction);
if(canProcessElements()) unfinishedInstructions.remove(instruction);
}
};
Processor<Iterable<String>, String> retriever0 = new AbstractRetrievalProcessor<Iterable<String>, String>(dataSourceType, String.class, executor, Collections.singleton(retriever1), 0) {
@Override
protected ProcessorInstruction<String> createInstruction(Iterable<String> element) {
AbstractProcessorInstruction<String> instruction = (AbstractProcessorInstruction<String>) super.createInstruction(element);
StatefulProcessorInstruction<String> statefulInstruction = new StatefulInstructionWrapper<>(instruction);
unfinishedInstructions.add(statefulInstruction);
return statefulInstruction;
}
@Override
protected Iterable<String> retrieveData(Iterable<String> element) {
logExecution("Retrieving data from data source");
return element;
}
@Override
public void instructionSucceeded(String result) {
logExecution("Data source retrieval finished");
super.instructionSucceeded(result);
}
@Override
public void afterInstructionFinished(ProcessorInstruction<String> instruction) {
super.afterInstructionFinished(instruction);
if(canProcessElements()) unfinishedInstructions.remove(instruction);
}
};
processors.add(retriever0);
processors.add(retriever1);
processors.add(heavyLoadProcessor);
processors.add(grouper);
processors.add(aggregator);
return retriever0;
}
};
}
private static class HeavyLoadInstruction extends StatefulBlockingInstruction<Element> {
private final Consumer<String> recorder;
public HeavyLoadInstruction(ProcessorInstructionHandler<Element> handler,
ProcessorInstructionPriority priority, long blockDuration, Element result, Consumer<String> recorder) {
super(handler, priority, blockDuration, result);
this.recorder = recorder;
}
@Override
public void run() {
recorder.accept("Executing heavy load instruction for " + result);
super.run();
}
@Override
protected void actionBeforeBlock() {
recorder.accept("Starting work for heavy load instruction for " + result);
}
@Override
protected void actionAfterBlock() {
recorder.accept("Finished heavy load instruction for " + result);
}
}
private static class StatefulInstructionWrapper<ResultType> extends StatefulProcessorInstruction<ResultType> {
private static Method computeResult;
private final AbstractProcessorInstruction<ResultType> instruction;
public StatefulInstructionWrapper(AbstractProcessorInstruction<ResultType> instruction) {
super(instruction.getHandler(), instruction.getPriority());
this.instruction = instruction;
}
@Override
@SuppressWarnings("unchecked")
protected ResultType internalComputeResult() throws Exception {
if (computeResult == null) {
computeResult = AbstractProcessorInstruction.class.getDeclaredMethod("computeResult");
computeResult.setAccessible(true);
}
return (ResultType) computeResult.invoke(instruction);
}
}
@After
public void resetComputeResultAccessibility() throws SecurityException, NoSuchMethodException {
AbstractProcessorInstruction.class.getDeclaredMethod("computeResult").setAccessible(false);
}
/**
* Simple data type consisting of a name and value.
*/
private static class Element {
private final String name;
private final int value;
public Element(String name, int value) {
this.name = name;
this.value = value;
}
public String getName() {
return name;
}
public int getValue() {
return value;
}
@Override
public String toString() {
return getName() + "-" + getValue();
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + ((name == null) ? 0 : name.hashCode());
result = prime * result + value;
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (obj == null)
return false;
if (getClass() != obj.getClass())
return false;
Element other = (Element) obj;
if (name == null) {
if (other.name != null)
return false;
} else if (!name.equals(other.name))
return false;
if (value != other.value)
return false;
return true;
}
}
}
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestProcessorQuery.java
+5 -5
View File
@@ -63,7 +63,7 @@ public class TestProcessorQuery {
Collection<Processor<Double, ?>> resultReceivers = new ArrayList<>();
resultReceivers.add(new AbortResultReceiver(resultReceiver));
return new BlockingProcessor<Iterable<Number>, Double>((Class<Iterable<Number>>)(Class<?>) Iterable.class, Double.class,
ConcurrencyTestsUtil.getExecutor(), resultReceivers, 1000) {
ConcurrencyTestsUtil.getSharedExecutor(), resultReceivers, 1000) {
@Override
protected Double createResult(Iterable<Number> element) {
return 0.0;
@@ -101,7 +101,7 @@ public class TestProcessorQuery {
Collection<Processor<Double, ?>> resultReceivers = new ArrayList<>();
resultReceivers.add(new AbortResultReceiver(resultReceiver));
return new BlockingProcessor<Iterable<Number>, Double>((Class<Iterable<Number>>)(Class<?>) Iterable.class, Double.class,
ConcurrencyTestsUtil.getExecutor(), resultReceivers, 1000) {
ConcurrencyTestsUtil.getSharedExecutor(), resultReceivers, 1000) {
@Override
protected Double createResult(Iterable<Number> element) {
return 0.0;
@@ -168,7 +168,7 @@ public class TestProcessorQuery {
resultReceivers.add(resultReceiver);
return new AbstractParallelProcessor<Iterable<Number>, Map<GroupKey, Double>>((Class<Iterable<Number>>)(Class<?>) Iterable.class,
(Class<Map<GroupKey, Double>>)(Class<?>) Map.class,
ConcurrencyTestsUtil.getExecutor(),
ConcurrencyTestsUtil.getSharedExecutor(),
resultReceivers) {
@Override
protected ProcessorInstruction<Map<GroupKey, Double>> createInstruction(final Iterable<Number> element) {
@@ -242,7 +242,7 @@ public class TestProcessorQuery {
resultReceivers.add(resultReceiver);
return new AbstractParallelProcessor<Double, Map<GroupKey, Double>>(Double.class,
(Class<Map<GroupKey, Double>>)(Class<?>) Map.class,
ConcurrencyTestsUtil.getExecutor(),
ConcurrencyTestsUtil.getSharedExecutor(),
resultReceivers) {
@Override
protected ProcessorInstruction<Map<GroupKey, Double>> createInstruction(Double element) {
@@ -282,7 +282,7 @@ public class TestProcessorQuery {
resultReceivers.add(resultReceiver);
return new AbstractParallelProcessor<Double, Map<GroupKey, Double>>(Double.class,
(Class<Map<GroupKey, Double>>)(Class<?>) Map.class,
ConcurrencyTestsUtil.getExecutor(),
ConcurrencyTestsUtil.getSharedExecutor(),
resultReceivers) {
@Override
protected ProcessorInstruction<Map<GroupKey, Double>> createInstruction(Double element) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractParallelProcessorElementProcessing.java
+124
View File
@@ -0,0 +1,124 @@
package com.sap.sse.datamining.impl.components;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.components.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.components.ProcessorInstruction;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
import com.sap.sse.datamining.test.util.components.StatefulBlockingInstruction;
public class TestAbstractParallelProcessorElementProcessing {
private ThreadPoolExecutor executor;
private Processor<Integer, Object> processor;
private List<StatefulBlockingInstruction<?>> createdInstructions;
@Before
public void initialize() {
int corePoolSize = Runtime.getRuntime().availableProcessors();
executor = new ThreadPoolExecutor(corePoolSize, corePoolSize, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
createdInstructions = new ArrayList<>();
processor = new AbstractParallelProcessor<Integer, Object>(Integer.class, Object.class, executor, Collections.emptySet()) {
@Override
protected ProcessorInstruction<Object> createInstruction(Integer sleepTime) {
StatefulBlockingInstruction<Object> instruction = new StatefulBlockingInstruction<>(this, sleepTime);
createdInstructions.add(instruction);
return instruction;
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { }
};
}
@Test
public void testSimpleProcessing() throws InterruptedException {
int elementCount = executor.getMaximumPoolSize() * 2;
for (int i = 0; i < elementCount; i++) {
processor.processElement(10);
}
assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
executor.shutdown();
assertThat("Executor couldn't terminate", executor.awaitTermination(1, TimeUnit.SECONDS), is(true));
for (StatefulBlockingInstruction<?> instruction : createdInstructions) {
assertThat("run wasn't called", instruction.runWasCalled(), is(true));
assertThat("computeResult wasn't called", instruction.computeResultWasCalled(), is(true));
assertThat("computeResult didn't finish", instruction.computeResultWasFinished(), is(true));
}
}
@Test
public void testProcessingAfterFinish() throws InterruptedException {
int instructionDuration = 50;
int elementCount = executor.getMaximumPoolSize() + 1; // Last instruction will be queued
for (int i = 0; i < elementCount; i++) {
processor.processElement(instructionDuration);
}
assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
Thread.sleep(instructionDuration / 3); // Giving some time to ensure execution of unqueued instructions
Thread finishingThread = ConcurrencyTestsUtil.tryToFinishTheProcessorInAnotherThread(processor);
do {
Thread.sleep(1);
} while (!finishingThread.isAlive());
assertThat("Processor is already finished", processor.isFinished(), is(false));
// Processor not yet finished. New elements will be accepted
processor.processElement(instructionDuration);
elementCount++;
assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
finishingThread.join(1000);
assertThat("Processor isn't finished", processor.isFinished(), is(true));
processor.processElement(instructionDuration);
assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
for (StatefulBlockingInstruction<?> instruction : createdInstructions) {
assertThat("run wasn't called", instruction.runWasCalled(), is(true));
assertThat("computeResult wasn't called", instruction.computeResultWasCalled(), is(true));
assertThat("computeResult didn't finish", instruction.computeResultWasFinished(), is(true));
}
}
@Test
public void testProcessingAfterAbort() throws InterruptedException {
int instructionDuration = 50;
int elementCount = executor.getMaximumPoolSize() + 1; // Last instruction will be queued
for (int i = 0; i < elementCount; i++) {
processor.processElement(instructionDuration);
}
assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
Thread.sleep(instructionDuration / 3); // Giving some time to ensure execution of unqueued instructions
processor.abort();
processor.processElement(0);
assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
executor.shutdown();
assertThat("Executor couldn't terminate", executor.awaitTermination(1, TimeUnit.SECONDS), is(true));
for (int i = 0; i < createdInstructions.size(); i++) {
StatefulBlockingInstruction<?> instruction = createdInstructions.get(i);
assertThat("run wasn't called", instruction.runWasCalled(), is(true));
// Last instructions was executed after the processor has been aborted. computeResult() should not be called
if (i == createdInstructions.size() - 1) {
assertThat("computeResult of last instruction was called", instruction.computeResultWasCalled(), is(false));
assertThat("computeResult of last instruction finished", instruction.computeResultWasFinished(), is(false));
} else {
assertThat("computeResult wasn't called", instruction.computeResultWasCalled(), is(true));
assertThat("computeResult didn't finish", instruction.computeResultWasFinished(), is(true));
}
}
}
}
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractParallelProcessorFinishing.java
+1 -1
View File
@@ -68,7 +68,7 @@ public class TestAbstractParallelProcessorFinishing {
}
private AbstractParallelProcessor<Integer, Integer> createProcessor(Collection<Processor<Integer, ?>> receivers) {
return new AbstractParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers) {
return new AbstractParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getSharedExecutor(), receivers) {
@Override
protected ProcessorInstruction<Integer> createInstruction(Integer partialElement) {
return new AbstractProcessorInstruction<Integer>(this) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractParallelProcessorWithManySimpleInstructions.java
+1 -1
View File
@@ -37,7 +37,7 @@ public class TestAbstractParallelProcessorWithManySimpleInstructions {
Collection<Processor<Integer, ?>> receivers = new ArrayList<>();
receivers.add(receiver);
processor = new AbstractParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers) {
processor = new AbstractParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getSharedExecutor(), receivers) {
@Override
protected ProcessorInstruction<Integer> createInstruction(final Integer element) {
return new AbstractProcessorInstruction<Integer>(this) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestDataRetrieverChainCreation.java
+11 -11
View File
@@ -1,6 +1,6 @@
package com.sap.sse.datamining.impl.components;
import static com.sap.sse.datamining.test.util.ConcurrencyTestsUtil.getExecutor;
import static com.sap.sse.datamining.test.util.ConcurrencyTestsUtil.getSharedExecutor;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.junit.Assert.assertThat;
@@ -90,12 +90,12 @@ public class TestDataRetrieverChainCreation {
assertThat(chainClone, not(dataRetrieverChainDefinition));
chainClone.startBuilding(ConcurrencyTestsUtil.getExecutor());
chainClone.startBuilding(ConcurrencyTestsUtil.getSharedExecutor());
}
@Test
public void testThatTheDataRetrieverChainBuilderHasToBeInitialized() {
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getExecutor());
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getSharedExecutor());
try {
chainBuilder.getCurrentRetrievedDataType();
@@ -126,7 +126,7 @@ public class TestDataRetrieverChainCreation {
@Test
public void testStepByStepDataRetrieverChainCreation() throws InterruptedException {
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getExecutor());
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getSharedExecutor());
chainBuilder.stepFurther(); //Initialization
assertThat(chainBuilder.getCurrentRetrievedDataType().equals(Test_Regatta.class), is(true));
@@ -206,7 +206,7 @@ public class TestDataRetrieverChainCreation {
@Test(expected=IllegalArgumentException.class)
public void testSettingAFilterWithWrongElementType() {
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getExecutor());
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getSharedExecutor());
chainBuilder.stepFurther(); //Initialization
chainBuilder.setFilter(raceFilter);
@@ -214,7 +214,7 @@ public class TestDataRetrieverChainCreation {
@Test(expected=IllegalArgumentException.class)
public void testSettingAResultReceiverWithWrongInputType() {
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getExecutor());
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getSharedExecutor());
chainBuilder.stepFurther(); //Initialization
chainBuilder.addResultReceiver(legReceiver);
@@ -253,12 +253,12 @@ public class TestDataRetrieverChainCreation {
raceRetrieverClass,
Test_HasRaceContext.class, "race");
dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getExecutor());
dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getSharedExecutor());
}
@Test(expected=IllegalStateException.class)
public void testSteppingToFar() {
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getExecutor());
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getSharedExecutor());
while (chainBuilder.canStepFurther()) {
chainBuilder.stepFurther();
}
@@ -272,7 +272,7 @@ public class TestDataRetrieverChainCreation {
chainWithSettings.endWith(TestLegOfCompetitorWithContextRetrievalProcessor.class, Test_RetrievalProcessorWithSettings.class, Test_HasLegOfCompetitorContext.class,
Test_RetrievalProcessorSettings.class, new Test_RetrievalProcessorSettings("Default Settings"), "legOfCompetitor");
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = chainWithSettings.startBuilding(getExecutor());
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = chainWithSettings.startBuilding(getSharedExecutor());
while (chainBuilder.canStepFurther()) {
chainBuilder.stepFurther();
}
@@ -311,7 +311,7 @@ public class TestDataRetrieverChainCreation {
chainWithSettings.endWith(TestLegOfCompetitorWithContextRetrievalProcessor.class, Test_RetrievalProcessorWithSettings.class, Test_HasLegOfCompetitorContext.class,
Test_RetrievalProcessorSettings.class, new Test_RetrievalProcessorSettings("Default Settings"), "legOfCompetitor");
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = chainWithSettings.startBuilding(getExecutor());
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = chainWithSettings.startBuilding(getSharedExecutor());
while (chainBuilder.canStepFurther()) {
chainBuilder.stepFurther();
}
@@ -320,7 +320,7 @@ public class TestDataRetrieverChainCreation {
@Test(expected=IllegalStateException.class)
public void testRetrieverWithNoSettingsButSettedSettingsCreated() {
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getExecutor());
DataRetrieverChainBuilder<Collection<Test_Regatta>> chainBuilder = dataRetrieverChainDefinition.startBuilding(ConcurrencyTestsUtil.getSharedExecutor());
chainBuilder.stepFurther(); // Initialization
chainBuilder.setSettings(new WrongSettings());
}
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestFilteringProcessors.java
+1 -1
View File
@@ -34,7 +34,7 @@ public class TestFilteringProcessors {
return element % 2 == 0;
}
};
Processor<Integer, Integer> filteringProcessor = new ParallelFilteringProcessor<Integer>(Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers, elementIsEvenCriteria);
Processor<Integer, Integer> filteringProcessor = new ParallelFilteringProcessor<Integer>(Integer.class, ConcurrencyTestsUtil.getSharedExecutor(), receivers, elementIsEvenCriteria);
ConcurrencyTestsUtil.processElements(filteringProcessor, createElementsToProcess());
ConcurrencyTestsUtil.sleepFor(100); // Giving the processor time to process the instructions
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestParallelExtractionProcessor.java
+2 -2
View File
@@ -63,7 +63,7 @@ public class TestParallelExtractionProcessor {
@Test
public void testValueExtraction() {
Processor<GroupedDataEntry<Number>, GroupedDataEntry<Integer>> processor = new ParallelGroupedElementsValueExtractionProcessor<Number, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, getCrossSumFunction);
Processor<GroupedDataEntry<Number>, GroupedDataEntry<Integer>> processor = new ParallelGroupedElementsValueExtractionProcessor<Number, Integer>(ConcurrencyTestsUtil.getSharedExecutor(), receivers, getCrossSumFunction);
Collection<GroupedDataEntry<Number>> elements = createElements();
ConcurrencyTestsUtil.processElements(processor, elements);
ConcurrencyTestsUtil.sleepFor(100); //Giving the processor time to finish the instructions
@@ -81,7 +81,7 @@ public class TestParallelExtractionProcessor {
@Test
public void testValueExtractionWithInvalidFunction() {
Processor<GroupedDataEntry<Number>, GroupedDataEntry<Integer>> processor = new ParallelGroupedElementsValueExtractionProcessor<Number, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, invalidFunction);
Processor<GroupedDataEntry<Number>, GroupedDataEntry<Integer>> processor = new ParallelGroupedElementsValueExtractionProcessor<Number, Integer>(ConcurrencyTestsUtil.getSharedExecutor(), receivers, invalidFunction);
ConcurrencyTestsUtil.processElements(processor, createElements());
ConcurrencyTestsUtil.sleepFor(100); //Giving the processor time to finish the instructions
assertThat("Values have been received, but the processor function is invalid.", receivedValues.isEmpty(), is(true));
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestParallelMultiDimensionalGroupingProcessor.java
+1 -1
View File
@@ -57,7 +57,7 @@ public class TestParallelMultiDimensionalGroupingProcessor {
@Test(expected=IllegalArgumentException.class)
public void testConstructionWithNullDimensions() {
new ParallelMultiDimensionsValueNestingGroupingProcessor<>(Number.class, ConcurrencyTestsUtil.getExecutor(), receivers, null);
new ParallelMultiDimensionsValueNestingGroupingProcessor<>(Number.class, ConcurrencyTestsUtil.getSharedExecutor(), receivers, null);
}
@Test(expected=IllegalArgumentException.class)
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestRetrieverFilterProcessorChain.java
+3 -3
View File
@@ -72,7 +72,7 @@ public class TestRetrieverFilterProcessorChain {
resultReceivers.add(retrievalProcessor);
@SuppressWarnings("unchecked")
Processor<Iterable<Iterable<Integer>>, Iterable<Integer>> layeredRetrievalProcessor = new AbstractRetrievalProcessor<Iterable<Iterable<Integer>>, Iterable<Integer>>((Class<Iterable<Iterable<Integer>>>)(Class<?>) Iterable.class, (Class<Iterable<Integer>>)(Class<?>) Iterable.class,
ConcurrencyTestsUtil.getExecutor(), resultReceivers, 0) {
ConcurrencyTestsUtil.getSharedExecutor(), resultReceivers, 0) {
@Override
protected Iterable<Iterable<Integer>> retrieveData(Iterable<Iterable<Integer>> element) {
return element;
@@ -110,11 +110,11 @@ public class TestRetrieverFilterProcessorChain {
return element >= 0;
}
};
Processor<Integer, Integer> filtrationProcessor = new ParallelFilteringProcessor<>(Integer.class, ConcurrencyTestsUtil.getExecutor(), filtrationResultReceivers, elementGreaterZeroFilterCriteria);
Processor<Integer, Integer> filtrationProcessor = new ParallelFilteringProcessor<>(Integer.class, ConcurrencyTestsUtil.getSharedExecutor(), filtrationResultReceivers, elementGreaterZeroFilterCriteria);
Collection<Processor<Integer, ?>> retrievalResultReceivers = new ArrayList<>();
retrievalResultReceivers.add(filtrationProcessor);
retrievalProcessor = new AbstractRetrievalProcessor<Iterable<Integer>, Integer>((Class<Iterable<Integer>>)(Class<?>) Iterable.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers, 1) {
retrievalProcessor = new AbstractRetrievalProcessor<Iterable<Integer>, Integer>((Class<Iterable<Integer>>)(Class<?>) Iterable.class, Integer.class, ConcurrencyTestsUtil.getSharedExecutor(), retrievalResultReceivers, 1) {
@Override
protected Iterable<Integer> retrieveData(Iterable<Integer> element) {
return element;
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestAbstractStoringParallelAggregationProcessor.java
+2 -2
View File
@@ -48,7 +48,7 @@ public class TestAbstractStoringParallelAggregationProcessor {
@Test
public void testAbstractAggregationHandling() throws InterruptedException {
Processor<GroupedDataEntry<Integer>, Map<GroupKey, Integer>> processor = new AbstractParallelGroupedDataStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") {
Processor<GroupedDataEntry<Integer>, Map<GroupKey, Integer>> processor = new AbstractParallelGroupedDataStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getSharedExecutor(), receivers, "Sum") {
@Override
protected void storeElement(GroupedDataEntry<Integer> element) {
elementStore.add(element);
@@ -104,7 +104,7 @@ public class TestAbstractStoringParallelAggregationProcessor {
}
}
});
Processor<GroupedDataEntry<Integer>, Map<GroupKey, Integer>> processor = new AbstractParallelGroupedDataStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") {
Processor<GroupedDataEntry<Integer>, Map<GroupKey, Integer>> processor = new AbstractParallelGroupedDataStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getSharedExecutor(), receivers, "Sum") {
@Override
protected void storeElement(GroupedDataEntry<Integer> element) {
if (element.getDataEntry() < 0) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelAggregationProcessors.java
+5 -5
View File
@@ -16,7 +16,7 @@ public class TestParallelAggregationProcessors extends AbstractTestParallelAvera
@Test
public void testSumAggregationProcessor() throws InterruptedException {
Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> sumAggregationProcessor = ParallelGroupedNumberDataSumAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> sumAggregationProcessor = ParallelGroupedNumberDataSumAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getSharedExecutor(), receivers);
Collection<GroupedDataEntry<Number>> elements = createElements();
ConcurrencyTestsUtil.processElements(sumAggregationProcessor, elements);
@@ -27,7 +27,7 @@ public class TestParallelAggregationProcessors extends AbstractTestParallelAvera
@Test
public void testMedianAggregationProcessor() throws InterruptedException {
Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> medianAggregationProcessor = ParallelGroupedNumberDataMedianAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> medianAggregationProcessor = ParallelGroupedNumberDataMedianAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getSharedExecutor(), receivers);
Collection<GroupedDataEntry<Number>> elements = createElements();
ConcurrencyTestsUtil.processElements(medianAggregationProcessor, elements);
@@ -38,7 +38,7 @@ public class TestParallelAggregationProcessors extends AbstractTestParallelAvera
@Test
public void testMaxAggregationProcessor() throws InterruptedException {
Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> maxAggregationProcessor = ParallelGroupedNumberDataMaxAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> maxAggregationProcessor = ParallelGroupedNumberDataMaxAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getSharedExecutor(), receivers);
Collection<GroupedDataEntry<Number>> elements = createElements();
ConcurrencyTestsUtil.processElements(maxAggregationProcessor, elements);
@@ -49,7 +49,7 @@ public class TestParallelAggregationProcessors extends AbstractTestParallelAvera
@Test
public void testMinAggregationProcessor() throws InterruptedException {
Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> minAggregationProcessor = ParallelGroupedNumberDataMinAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> minAggregationProcessor = ParallelGroupedNumberDataMinAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getSharedExecutor(), receivers);
Collection<GroupedDataEntry<Number>> elements = createElements();
ConcurrencyTestsUtil.processElements(minAggregationProcessor, elements);
@@ -60,7 +60,7 @@ public class TestParallelAggregationProcessors extends AbstractTestParallelAvera
@Test
public void testCountAggregationProcessor() throws InterruptedException {
Processor<GroupedDataEntry<Object>, Map<GroupKey, Number>> countAggregationProcessor = ParallelGroupedDataCountAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
Processor<GroupedDataEntry<Object>, Map<GroupKey, Number>> countAggregationProcessor = ParallelGroupedDataCountAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getSharedExecutor(), receivers);
@SuppressWarnings("unchecked")
Collection<GroupedDataEntry<Object>> elements = (Collection<GroupedDataEntry<Object>>)(Collection<?>) createElements();
ConcurrencyTestsUtil.processElements(countAggregationProcessor, elements);
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelAveragingProcessors.java
+1 -1
View File
@@ -19,7 +19,7 @@ public class TestParallelAveragingProcessors extends AbstractTestParallelAveragi
@Test
public void testAverageAggregationProcessor() throws InterruptedException {
Processor<GroupedDataEntry<Number>, Map<GroupKey, AverageWithStats<Number>>> averageAggregationProcessor = ParallelGroupedNumberDataAverageAggregationProcessor
.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
.getDefinition().construct(ConcurrencyTestsUtil.getSharedExecutor(), receivers);
Collection<GroupedDataEntry<Number>> elements = createElements();
ConcurrencyTestsUtil.processElements(averageAggregationProcessor, elements);
averageAggregationProcessor.finish();
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelGroupedDataCollectingAsSetProcessor.java
+1 -1
View File
@@ -37,7 +37,7 @@ public class TestParallelGroupedDataCollectingAsSetProcessor {
@Test
public void testDataCollecting() throws InterruptedException {
Processor<GroupedDataEntry<Double>, Map<GroupKey, HashSet<Double>>> collectingProcessor = new ParallelGroupedDataCollectingAsSetProcessor<Double>(ConcurrencyTestsUtil.getExecutor(), receivers);
Processor<GroupedDataEntry<Double>, Map<GroupKey, HashSet<Double>>> collectingProcessor = new ParallelGroupedDataCollectingAsSetProcessor<Double>(ConcurrencyTestsUtil.getSharedExecutor(), receivers);
Collection<GroupedDataEntry<Double>> elements = createElements();
ConcurrencyTestsUtil.processElements(collectingProcessor, elements);
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/functions/TestFunctionManagerAsFunctionRegistry.java
+1 -1
View File
@@ -63,7 +63,7 @@ public class TestFunctionManagerAsFunctionRegistry {
DataRetrieverChainDefinitionRegistry dataRetrieverChainDefinitionRegistry = new DataRetrieverChainDefinitionManager();
AggregationProcessorDefinitionRegistry aggregationProcessorDefinitionRegistry = new AggregationProcessorDefinitionManager();
QueryDefinitionDTORegistry queryDefinitionRegistry = new QueryDefinitionDTOManager();
ModifiableDataMiningServer server = new DataMiningServerImpl(ConcurrencyTestsUtil.getExecutor(), functionManager,
ModifiableDataMiningServer server = new DataMiningServerImpl(ConcurrencyTestsUtil.getSharedExecutor(), functionManager,
dataSourceProviderRegistry,
dataRetrieverChainDefinitionRegistry,
aggregationProcessorDefinitionRegistry,
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/ComponentTestsUtil.java
+1 -1
View File
@@ -22,7 +22,7 @@ import com.sap.sse.datamining.test.domain.impl.Test_TeamImpl;
public final class ComponentTestsUtil {
private final static ProcessorFactory processorFactory = new ProcessorFactory(ConcurrencyTestsUtil.getExecutor());
private final static ProcessorFactory processorFactory = new ProcessorFactory(ConcurrencyTestsUtil.getSharedExecutor());
public static ProcessorFactory getProcessorFactory() {
return processorFactory;
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/ConcurrencyTestsUtil.java
+3 -2
View File
@@ -19,7 +19,7 @@ public class ConcurrencyTestsUtil extends TestsUtil {
private static final int THREAD_POOL_SIZE = Math.max(Runtime.getRuntime().availableProcessors(), 3);
private static final ExecutorService executor = new DataMiningExecutorService(THREAD_POOL_SIZE);
public static ExecutorService getExecutor() {
public static ExecutorService getSharedExecutor() {
return executor;
}
@@ -47,7 +47,7 @@ public class ConcurrencyTestsUtil extends TestsUtil {
}
}
public static void tryToFinishTheProcessorInAnotherThread(final Processor<?, ?> processor) {
public static Thread tryToFinishTheProcessorInAnotherThread(final Processor<?, ?> processor) {
Thread finishingThread = new Thread(new Runnable() {
@Override
public void run() {
@@ -59,6 +59,7 @@ public class ConcurrencyTestsUtil extends TestsUtil {
}
});
finishingThread.start();
return finishingThread;
}
protected ConcurrencyTestsUtil() {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/TestsUtil.java
+6 -1
View File
@@ -1,6 +1,7 @@
package com.sap.sse.datamining.test.util;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.ModifiableDataMiningServer;
import com.sap.sse.datamining.components.management.AggregationProcessorDefinitionRegistry;
@@ -55,12 +56,16 @@ public class TestsUtil {
}
public static ModifiableDataMiningServer createNewServer() {
return createNewServer(ConcurrencyTestsUtil.getSharedExecutor());
}
public static ModifiableDataMiningServer createNewServer(ExecutorService executor) {
FunctionRegistry functionRegistry = new FunctionManager();
DataSourceProviderRegistry dataSourceProviderRegistry = new DataSourceProviderManager();
DataRetrieverChainDefinitionRegistry dataRetrieverChainDefinitionRegistry = new DataRetrieverChainDefinitionManager();
AggregationProcessorDefinitionRegistry aggregationProcessorDefinitionRegistry = new AggregationProcessorDefinitionManager();
QueryDefinitionDTORegistry queryDefinitionRegistry = new QueryDefinitionDTOManager();
return new DataMiningServerImpl(ConcurrencyTestsUtil.getExecutor(), functionRegistry,
return new DataMiningServerImpl(executor, functionRegistry,
dataSourceProviderRegistry,
dataRetrieverChainDefinitionRegistry,
aggregationProcessorDefinitionRegistry,
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/StatefulBlockingInstruction.java
+42
View File
@@ -0,0 +1,42 @@
package com.sap.sse.datamining.test.util.components;
import com.sap.sse.datamining.components.ProcessorInstructionHandler;
import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
public class StatefulBlockingInstruction<ResultType> extends StatefulProcessorInstruction<ResultType> {
protected final long blockDuration;
protected final ResultType result;
public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, long blockDuration) {
this(handler, 0, blockDuration, null);
}
public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, ProcessorInstructionPriority priority, long blockDuration, ResultType result) {
this(handler, priority.asIntValue(), blockDuration, result);
}
public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, int priority, long blockDuration, ResultType result) {
super(handler, priority);
this.blockDuration = blockDuration;
this.result = result;
}
@Override
protected ResultType internalComputeResult() throws Exception {
actionBeforeBlock();
if (blockDuration > 0) {
Thread.sleep(blockDuration);
}
actionAfterBlock();
return result;
}
protected void actionBeforeBlock() { }
protected void actionAfterBlock() { }
public long getBlockDuration() {
return blockDuration;
}
}
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/StatefulProcessorInstruction.java
+54
View File
@@ -0,0 +1,54 @@
package com.sap.sse.datamining.test.util.components;
import com.sap.sse.datamining.components.ProcessorInstructionHandler;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
public abstract class StatefulProcessorInstruction<ResultType> extends AbstractProcessorInstruction<ResultType> {
private boolean runWasCalled;
private boolean computeResultWasCalled;
private boolean computeResultWasFinished;
public StatefulProcessorInstruction(ProcessorInstructionHandler<ResultType> processor, int priority) {
super(processor, priority);
}
public StatefulProcessorInstruction(ProcessorInstructionHandler<ResultType> handler,
ProcessorInstructionPriority priority) {
super(handler, priority);
}
public StatefulProcessorInstruction(ProcessorInstructionHandler<ResultType> handler) {
super(handler);
}
@Override
public void run() {
runWasCalled = true;
super.run();
}
@Override
protected ResultType computeResult() throws Exception {
computeResultWasCalled = true;
ResultType result = internalComputeResult();
computeResultWasFinished = true;
return result;
}
protected abstract ResultType internalComputeResult() throws Exception;
public boolean runWasCalled() {
return runWasCalled;
}
public boolean computeResultWasCalled() {
return computeResultWasCalled;
}
public boolean computeResultWasFinished() {
return computeResultWasFinished;
}
}
java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessorInstructionHandler.java
+1 -1
View File
@@ -7,6 +7,6 @@ public interface ProcessorInstructionHandler<ResultType> {
void instructionSucceeded(ResultType result);
void instructionFailed(Exception e);
void afterInstructionFinished();
void afterInstructionFinished(ProcessorInstruction<ResultType> instruction);
}
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java
+7 -4
View File
@@ -59,18 +59,21 @@ public abstract class AbstractParallelProcessor<InputType, ResultType> extends A
private boolean isInstructionValid(ProcessorInstruction<ResultType> instruction) {
return instruction != null;
}
@Override
public void instructionSucceeded(ResultType result) {
forwardResultToReceivers(result);
}
@Override
public void instructionFailed(Exception e) {
if (!isAborted() || !(e instanceof InterruptedException)) {
onFailure(e);
}
}
public void afterInstructionFinished() {
@Override
public void afterInstructionFinished(ProcessorInstruction<ResultType> instruction) {
unfinishedInstructionsCounter.getAndDecrement();
}
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractProcessorInstruction.java
+5 -1
View File
@@ -59,12 +59,16 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Proces
} catch (Exception e) {
handler.instructionFailed(e);
} finally {
handler.afterInstructionFinished();
handler.afterInstructionFinished(this);
}
}
protected abstract ResultType computeResult() throws Exception;
public ProcessorInstructionHandler<ResultType> getHandler() {
return handler;
}
@Override
public int getPriority() {
return priority;
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelAggregationProcessor.java
+2 -1
View File
@@ -5,6 +5,7 @@ import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.components.ProcessorInstruction;
import com.sap.sse.datamining.impl.components.AbstractParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
@@ -26,7 +27,7 @@ public abstract class AbstractParallelAggregationProcessor<InputType, Aggregated
}
@Override
protected AbstractProcessorInstruction<AggregatedType> createInstruction(final InputType element) {
protected ProcessorInstruction<AggregatedType> createInstruction(final InputType element) {
if (needsSynchronization()) {
return new AbstractProcessorInstruction<AggregatedType>(this, ProcessorInstructionPriority.Aggregation) {
@Override