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

Adjusted the tests for instructions that can abort their execution

Browse Source
This commit is contained in:
Lennart Hensler
2018-07-31 13:07:53 +02:00
parent fd1f79e802
commit 6b94358542
4 changed files with 125 additions and 61 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
+28 -12
View File
@@ -71,24 +71,31 @@ 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,
* retrieved for each group. This means that each group should have a runtime of {@value #HeavyLoadInstructionTotalDuration}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 time a step of a heavy load instruction blocks the executing thread (using {@link Thread#sleep(long)}). */
private static final long HeavyLoadInstructionStepDuration = 50;
/** The number of steps a heavy load instruction performs */
private static final int HeavyLoadInstructionNumberOfSteps = 10;
/** The total time a heavy load instruction blocks the executing thread (using {@link Thread#sleep(long)}). */
private static final long HeavyLoadInstructionTotalDuration = HeavyLoadInstructionStepDuration * HeavyLoadInstructionNumberOfSteps;
/** The number of milliseconds to wait before {@link Query#abort()} is called. */
private static final long AbortQueryDelay = (long) (HeavyLoadInstructionDuration * 5.45);
private static final long AbortQueryDelay = (long) (HeavyLoadInstructionTotalDuration * 5.45);
/** The time given to the executor to complete all unfinished instructions */
private static final long TerminationTimeout = (long) (HeavyLoadInstructionDuration * 1.5);
private static final long TerminationTimeout = HeavyLoadInstructionStepDuration * 2;
//------------------------------------------------------------------------------------------------------------
// 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 boolean RecordExecution = true;
private static final SimpleDateFormat DateFormatter = new SimpleDateFormat("HH:mm:ss.SSS");
private ConcurrentLinkedQueue<String> executionRecord;
//------------------------------------------------------------------------------------------------------------
@@ -174,6 +181,10 @@ public class TestAbortingHeavyLoadQuery {
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());
if (instruction instanceof StatefulBlockingInstruction) {
StatefulBlockingInstruction<?> blockingInstruction = (StatefulBlockingInstruction<?>) instruction;
assertTrue("computeResult() of a running heavy load instruction wasn't aborted", blockingInstruction.computeResultWasAborted());
}
}
for (StatefulProcessorInstruction<?> instruction : notStartedInstructions) {
assertTrue("run() of an unstarted unfinished instruction wasn't called", instruction.runWasCalled());
@@ -216,7 +227,7 @@ public class TestAbortingHeavyLoadQuery {
* </li>
* <li>
* A heavy load instruction for each element is scheduled, which blocks the running thread for
* {@value #HeavyLoadInstructionDuration}ms.
* {@value #HeavyLoadInstructionTotalDuration}ms.
* </li>
* <li>
* Each element is grouped by its name and its value is used as value for the {@link GroupedDataEntry}.
@@ -306,8 +317,8 @@ public class TestAbortingHeavyLoadQuery {
@Override
protected ProcessorInstruction<Element> createInstruction(Element element) {
StatefulProcessorInstruction<Element> instruction = new HeavyLoadInstruction(this,
ProcessorInstructionPriority.Extraction, HeavyLoadInstructionDuration, element,
TestAbortingHeavyLoadQuery.this::logExecution);
ProcessorInstructionPriority.Extraction, HeavyLoadInstructionStepDuration,
HeavyLoadInstructionNumberOfSteps, element, TestAbortingHeavyLoadQuery.this::logExecution);
unfinishedInstructions.add(instruction);
return instruction;
}
@@ -397,9 +408,9 @@ public class TestAbortingHeavyLoadQuery {
private final Consumer<String> recorder;
public HeavyLoadInstruction(ProcessorInstructionHandler<Element> handler,
ProcessorInstructionPriority priority, long blockDuration, Element result, Consumer<String> recorder) {
super(handler, priority, blockDuration, result);
public HeavyLoadInstruction(ProcessorInstructionHandler<Element> handler, ProcessorInstructionPriority priority,
long stepDuration, int numberOfSteps, Element result, Consumer<String> recorder) {
super(handler, priority, stepDuration, numberOfSteps, result);
this.recorder = recorder;
}
@@ -414,6 +425,11 @@ public class TestAbortingHeavyLoadQuery {
recorder.accept("Starting work for heavy load instruction for " + result);
}
@Override
protected void actionBeforeAbort() {
recorder.accept("Aborting heavy load instruction for " + result);
}
@Override
protected void actionAfterBlock() {
recorder.accept("Finished heavy load instruction for " + result);
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractParallelProcessorElementProcessing.java
+62 -34
View File
@@ -1,7 +1,9 @@
package com.sap.sse.datamining.impl.components;
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.util.ArrayList;
import java.util.Collections;
@@ -13,6 +15,7 @@ import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.common.Util.Pair;
import com.sap.sse.datamining.components.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.components.ProcessorInstruction;
@@ -22,7 +25,7 @@ import com.sap.sse.datamining.test.util.components.StatefulBlockingInstruction;
public class TestAbstractParallelProcessorElementProcessing {
private ThreadPoolExecutor executor;
private Processor<Integer, Object> processor;
private Processor<Pair<Long, Integer>, Object> processor;
private List<StatefulBlockingInstruction<?>> createdInstructions;
@Before
@@ -30,10 +33,13 @@ public class TestAbstractParallelProcessorElementProcessing {
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()) {
@SuppressWarnings("unchecked")
Class<Pair<Long, Integer>> inputType = (Class<Pair<Long, Integer>>)(Class<?>) Pair.class;
processor = new AbstractParallelProcessor<Pair<Long, Integer>, Object>(inputType, Object.class, executor, Collections.emptySet()) {
@Override
protected ProcessorInstruction<Object> createInstruction(Integer sleepTime) {
StatefulBlockingInstruction<Object> instruction = new StatefulBlockingInstruction<>(this, sleepTime);
protected ProcessorInstruction<Object> createInstruction(Pair<Long, Integer> sleepTime) {
StatefulBlockingInstruction<Object> instruction = new StatefulBlockingInstruction<>(this, sleepTime.getA(), sleepTime.getB());
createdInstructions.add(instruction);
return instruction;
}
@@ -44,79 +50,101 @@ public class TestAbstractParallelProcessorElementProcessing {
@Test
public void testSimpleProcessing() throws InterruptedException {
long stepDuration = 10;
int numberOfSteps = 1;
Pair<Long, Integer> input = new Pair<Long, Integer>(stepDuration, numberOfSteps);
int elementCount = executor.getMaximumPoolSize() * 2;
for (int i = 0; i < elementCount; i++) {
processor.processElement(10);
processor.processElement(input);
}
assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
double executionTime = Math.ceil((double) elementCount / executor.getMaximumPoolSize()) * stepDuration * numberOfSteps;
executor.shutdown();
assertThat("Executor couldn't terminate", executor.awaitTermination(1, TimeUnit.SECONDS), is(true));
assertTrue("Executor couldn't terminate", executor.awaitTermination((long) (executionTime * 1.2), TimeUnit.SECONDS));
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));
assertTrue("run wasn't called", instruction.runWasCalled());
assertTrue("computeResult wasn't called", instruction.computeResultWasCalled());
assertTrue("computeResult didn't finish", instruction.computeResultWasFinished());
assertFalse("computeResult was aborted", instruction.computeResultWasAborted());
}
}
@Test
public void testProcessingAfterFinish() throws InterruptedException {
int instructionDuration = 50;
long stepDuration = 10;
int numberOfSteps = 5;
Pair<Long, Integer> input = new Pair<Long, Integer>(stepDuration, numberOfSteps);
int elementCount = executor.getMaximumPoolSize() + 1; // Last instruction will be queued
for (int i = 0; i < elementCount; i++) {
processor.processElement(instructionDuration);
processor.processElement(input);
}
assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
Thread.sleep(instructionDuration / 3); // Giving some time to ensure execution of unqueued instructions
Thread.sleep(stepDuration); // 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));
assertFalse("Processor is already finished", processor.isFinished());
// Processor not yet finished. New elements will be accepted
processor.processElement(instructionDuration);
processor.processElement(input);
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);
finishingThread.join();
assertTrue("Processor isn't finished", processor.isFinished());
processor.processElement(input);
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));
assertTrue("run wasn't called", instruction.runWasCalled());
assertTrue("computeResult wasn't called", instruction.computeResultWasCalled());
assertTrue("computeResult didn't finish", instruction.computeResultWasFinished());
assertFalse("computeResult was aborted", instruction.computeResultWasAborted());
}
}
@Test
public void testProcessingAfterAbort() throws InterruptedException {
int instructionDuration = 50;
int elementCount = executor.getMaximumPoolSize() + 1; // Last instruction will be queued
long stepDuration = 10;
int numberOfSteps = 5;
Pair<Long, Integer> input = new Pair<Long, Integer>(stepDuration, numberOfSteps);
int elementCount = executor.getMaximumPoolSize() // Set of finished instructions
+ executor.getMaximumPoolSize() // Set of started, but not yet finished instructions
+ 1; // Scheduled instruction
for (int i = 0; i < elementCount; i++) {
processor.processElement(instructionDuration);
processor.processElement(input);
}
assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
Thread.sleep(instructionDuration / 3); // Giving some time to ensure execution of unqueued instructions
long instructionDuration = stepDuration * numberOfSteps;
Thread.sleep(instructionDuration + stepDuration); // Time to finish first set and start second set
processor.abort();
processor.processElement(0);
processor.processElement(new Pair<>(0L, 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));
assertTrue("Executor couldn't terminate", executor.awaitTermination(2 * stepDuration, TimeUnit.MILLISECONDS));
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));
// run should be called for all instructions
assertTrue("run wasn't called", instruction.runWasCalled());
if (i < executor.getMaximumPoolSize()) {
// First set of instructions should be processed normally
assertTrue("computeResult wasn't called", instruction.computeResultWasCalled());
assertTrue("computeResult didn't finish", instruction.computeResultWasFinished());
assertFalse("computeResult was aborted", instruction.computeResultWasAborted());
} else if (i < executor.getMaximumPoolSize() * 2) {
// Second set of instructions was running when the processor was aborted. computeResult should be aborted
assertTrue("computeResult wasn't called", instruction.computeResultWasCalled());
assertTrue("computeResult didn't finish", instruction.computeResultWasFinished());
assertTrue("computeResult wasn't aborted", instruction.computeResultWasAborted());
} else {
assertThat("computeResult wasn't called", instruction.computeResultWasCalled(), is(true));
assertThat("computeResult didn't finish", instruction.computeResultWasFinished(), is(true));
// Last instructions was still scheduled when the processor was aborted. computeResult should not be called
assertFalse("computeResult of last instruction was called", instruction.computeResultWasCalled());
assertFalse("computeResult of last instruction finished", instruction.computeResultWasFinished());
assertFalse("computeResult was aborted", instruction.computeResultWasAborted());
}
}
}
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/StatefulBlockingInstruction.java
+30 -14
View File
@@ -5,38 +5,54 @@ import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
public class StatefulBlockingInstruction<ResultType> extends StatefulProcessorInstruction<ResultType> {
protected final long blockDuration;
protected final long stepDuration;
protected final int numberOfSteps;
protected final ResultType result;
private boolean computeResultWasAborted;
public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, long blockDuration) {
this(handler, 0, blockDuration, null);
public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, long stepDuration, int numberOfSteps) {
this(handler, 0, stepDuration, numberOfSteps, null);
}
public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, ProcessorInstructionPriority priority, long blockDuration, ResultType result) {
this(handler, priority.asIntValue(), blockDuration, result);
public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, ProcessorInstructionPriority priority, long stepDuration, int numberOfSteps, ResultType result) {
this(handler, priority.asIntValue(), stepDuration, numberOfSteps, result);
}
public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, int priority, long blockDuration, ResultType result) {
public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, int priority, long stepDuration, int numberOfSteps, ResultType result) {
super(handler, priority);
this.blockDuration = blockDuration;
this.stepDuration = stepDuration;
this.numberOfSteps = numberOfSteps;
this.result = result;
}
@Override
protected ResultType internalComputeResult() throws Exception {
actionBeforeBlock();
if (blockDuration > 0) {
Thread.sleep(blockDuration);
if (getTotalBlockDuration() > 0) {
actionBeforeBlock();
for (int i = 0; i < numberOfSteps; i++) {
if (isAborted()) {
actionBeforeAbort();
computeResultWasAborted = true;
break;
}
Thread.sleep(stepDuration);
}
actionAfterBlock();
}
actionAfterBlock();
return result;
}
protected void actionBeforeBlock() { }
protected void actionBeforeAbort() { }
protected void actionAfterBlock() { }
public long getBlockDuration() {
return blockDuration;
public long getTotalBlockDuration() {
return stepDuration * numberOfSteps;
}
public boolean computeResultWasAborted() {
return computeResultWasAborted;
}
}
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractProcessorInstruction.java
+5 -1
View File
@@ -52,7 +52,7 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Proces
@Override
public void run() {
try {
if (!handler.isAborted()) {
if (!isAborted()) {
ResultType result = computeResult();
handler.instructionSucceeded(result);
}
@@ -62,6 +62,10 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Proces
handler.afterInstructionFinished(this);
}
}
protected boolean isAborted() {
return handler.isAborted();
}
protected abstract ResultType computeResult() throws Exception;