From 0698098f3502ddfb3f2b234ff7b0b493db2f612c Mon Sep 17 00:00:00 2001 From: Lennart Hensler Date: Sat, 15 Mar 2014 18:03:49 +0100 Subject: [PATCH] Refactored the filtering retrieval processer to fit the standard usage --- .../GPSFixFilteringRetrievalProcessor.java | 33 +++++++++ .../components/GPSFixRetrievalProcessor.java | 36 ---------- .../datamining/impl/TestProcessorQuery.java | 13 +--- ...stAbstractFilteringRetrievalProcessor.java | 12 +--- ...llelMultiDimensionalGroupingProcessor.java | 5 +- .../AbstractFilteringRetrievalProcessor.java | 67 +++++++++++++------ ...AbstractPartitioningParallelProcessor.java | 15 +++-- ...llelMultiDimensionalGroupingProcessor.java | 7 +- ...ctParallelStoringAggregationProcessor.java | 11 +-- 9 files changed, 99 insertions(+), 100 deletions(-) create mode 100644 java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixFilteringRetrievalProcessor.java delete mode 100644 java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java diff --git a/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixFilteringRetrievalProcessor.java b/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixFilteringRetrievalProcessor.java new file mode 100644 index 00000000000..0047481d960 --- /dev/null +++ b/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixFilteringRetrievalProcessor.java @@ -0,0 +1,33 @@ +package com.sap.sailing.datamining.impl.components; + +import java.util.Collection; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; + +import com.sap.sailing.domain.leaderboard.LeaderboardGroup; +import com.sap.sailing.server.RacingEventService; +import com.sap.sse.datamining.components.FilterCriteria; +import com.sap.sse.datamining.components.Processor; +import com.sap.sse.datamining.impl.components.AbstractFilteringRetrievalProcessor; + +public class GPSFixFilteringRetrievalProcessor extends + AbstractFilteringRetrievalProcessor { + + public GPSFixFilteringRetrievalProcessor(ExecutorService executor, + Collection> resultReceivers, FilterCriteria criteria) { + super(executor, resultReceivers, criteria); + } + + @Override + protected Iterable retrieveData(RacingEventService element) { + // TODO Auto-generated method stub + return null; + } + + @Override + protected Callable createInstruction(LeaderboardGroup filteredPartialElement) { + // TODO Auto-generated method stub + return null; + } + +} diff --git a/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java b/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java deleted file mode 100644 index b39a7f4a147..00000000000 --- a/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java +++ /dev/null @@ -1,36 +0,0 @@ -package com.sap.sailing.datamining.impl.components; - -import java.util.Collection; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutorService; - -import com.sap.sailing.datamining.data.GPSFixWithContext; -import com.sap.sailing.domain.leaderboard.LeaderboardGroup; -import com.sap.sailing.server.RacingEventService; -import com.sap.sse.datamining.AdditionalResultDataBuilder; -import com.sap.sse.datamining.components.Processor; -import com.sap.sse.datamining.impl.components.AbstractPartitioningParallelProcessor; - -public class GPSFixRetrievalProcessor extends - AbstractPartitioningParallelProcessor { - - public GPSFixRetrievalProcessor(ExecutorService executor, Collection> resultReceivers) { - super(executor, resultReceivers); - } - - @Override - protected Callable createInstruction(LeaderboardGroup partialElement) { - throw new UnsupportedOperationException(); - } - - @Override - protected Iterable partitionElement(RacingEventService element) { - throw new UnsupportedOperationException(); - } - - @Override - protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { - throw new UnsupportedOperationException(); - } - -} diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestProcessorQuery.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestProcessorQuery.java index ac9a7b0a1df..f3e9e24fb30 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestProcessorQuery.java +++ b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestProcessorQuery.java @@ -114,7 +114,7 @@ public class TestProcessorQuery { Collection> dimensions = new ArrayList<>(); Function getLengthFunction = FunctionFactory.createMethodWrappingFunction(FunctionTestsUtil.getMethodFromClass(Number.class, "getLength")); dimensions.add(getLengthFunction); - Processor> lengthGrouper = new ParallelMultiDimensionalGroupingProcessor<>(executor, Arrays.asList(crossSumExtractor), dimensions); + Processor lengthGrouper = new ParallelMultiDimensionalGroupingProcessor<>(executor, Arrays.asList(crossSumExtractor), dimensions); FilterCriteria retrievalFilterCriteria = new FilterCriteria() { @Override @@ -122,20 +122,11 @@ public class TestProcessorQuery { return element.getValue() >= 10; } }; - Processor> filteringRetrievalProcessor = new AbstractFilteringRetrievalProcessor, Number, Iterable>(ConcurrencyTestsUtil.getExecutor(), Arrays.asList(lengthGrouper), retrievalFilterCriteria) { + Processor> filteringRetrievalProcessor = new AbstractFilteringRetrievalProcessor, Number>(ConcurrencyTestsUtil.getExecutor(), Arrays.asList(lengthGrouper), retrievalFilterCriteria) { @Override protected Iterable retrieveData(Iterable element) { return element; } - @Override - protected Callable> createInstruction(final Number filteredPartialElement) { - return new Callable>() { - @Override - public Iterable call() throws Exception { - return Arrays.asList(filteredPartialElement); - } - }; - } }; query.setFirstProcessor(filteringRetrievalProcessor); diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractFilteringRetrievalProcessor.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractFilteringRetrievalProcessor.java index 3335fd81060..b27f636411a 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractFilteringRetrievalProcessor.java +++ b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractFilteringRetrievalProcessor.java @@ -6,7 +6,6 @@ import static org.junit.Assert.assertThat; import java.util.Arrays; import java.util.Collection; import java.util.HashSet; -import java.util.concurrent.Callable; import org.junit.Before; import org.junit.Test; @@ -75,20 +74,11 @@ public class TestAbstractFilteringRetrievalProcessor { return element >= 0; } }; - filteringRetrievalProcessor = new AbstractFilteringRetrievalProcessor, Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), Arrays.asList(resultReceiver), filterCriteria) { + filteringRetrievalProcessor = new AbstractFilteringRetrievalProcessor, Integer>(ConcurrencyTestsUtil.getExecutor(), Arrays.asList(resultReceiver), filterCriteria) { @Override protected Iterable retrieveData(Iterable element) { return element; } - @Override - protected Callable createInstruction(final Integer filteredPartialElement) { - return new Callable() { - @Override - public Integer call() throws Exception { - return filteredPartialElement; - } - }; - } }; } diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestParallelMultiDimensionalGroupingProcessor.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestParallelMultiDimensionalGroupingProcessor.java index 2b5a79770b6..14090d8f29a 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestParallelMultiDimensionalGroupingProcessor.java +++ b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestParallelMultiDimensionalGroupingProcessor.java @@ -5,7 +5,6 @@ import static org.junit.Assert.assertThat; import java.lang.reflect.Method; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; import org.junit.Before; @@ -25,7 +24,7 @@ import com.sap.sse.datamining.test.util.FunctionTestsUtil; public class TestParallelMultiDimensionalGroupingProcessor { - private Processor> processor; + private Processor processor; private Collection>> receivers; private GroupedDataEntry groupedElement; @@ -79,7 +78,7 @@ public class TestParallelMultiDimensionalGroupingProcessor { @Test public void testGroupKeyGeneration() { Number number = new Number(1111); - processor.onElement(Arrays.asList(number)); + processor.onElement(number); ConcurrencyTestsUtil.sleepFor(200); //Giving the processor time to finish the instruction verifyGroupedElement(number); } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractFilteringRetrievalProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractFilteringRetrievalProcessor.java index 8fc5ded98e1..884173cfb71 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractFilteringRetrievalProcessor.java +++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractFilteringRetrievalProcessor.java @@ -1,48 +1,73 @@ package com.sap.sse.datamining.impl.components; -import java.util.ArrayList; import java.util.Collection; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import com.sap.sse.datamining.AdditionalResultDataBuilder; import com.sap.sse.datamining.components.FilterCriteria; import com.sap.sse.datamining.components.Processor; -public abstract class AbstractFilteringRetrievalProcessor - extends AbstractPartitioningParallelProcessor { +public abstract class AbstractFilteringRetrievalProcessor + extends AbstractPartitioningParallelProcessor { - private final FilterCriteria criteria; + private final FilterCriteria criteria; + private Lock retrievedDataAmountLock; private int retrievedDataAmount; + + private Lock filteredDataAmountLock; private int filteredDataAmount; - public AbstractFilteringRetrievalProcessor(ExecutorService executor, Collection> resultReceivers, FilterCriteria criteria) { + public AbstractFilteringRetrievalProcessor(ExecutorService executor, + Collection> resultReceivers, FilterCriteria criteria) { super(executor, resultReceivers); this.criteria = criteria; + retrievedDataAmountLock = new ReentrantLock(); + filteredDataAmountLock = new ReentrantLock(); } @Override - protected Iterable partitionElement(InputType element) { - Collection filteredData = new ArrayList<>(); - - Iterable retrievedData = retrieveData(element); - for (WorkingType retrievedDataEntry : retrievedData) { - retrievedDataAmount++; - if (criteria.matches(retrievedDataEntry)) { - filteredData.add(retrievedDataEntry); - } - } - filteredDataAmount = filteredData.size(); - return filteredData; + protected Iterable partitionElement(InputType element) { + return retrieveData(element); } - protected abstract Iterable retrieveData(InputType element); + protected abstract Iterable retrieveData(InputType element); - //Redefinition of the method to set the parameter name to element instead of partial element. - //This makes the implementation of sub classes more fluent. @Override - protected abstract Callable createInstruction(WorkingType filteredPartialElement); + protected Callable createInstruction(final ResultType partialElement) { + return new Callable() { + @Override + public ResultType call() throws Exception { + incrementRetrievedDataAmount(); + if (criteria.matches(partialElement)) { + incrementFilteredDataAmount(); + return partialElement; + } + return createInvalidResult(); + } + }; + } + + private void incrementRetrievedDataAmount() { + retrievedDataAmountLock.lock(); + try { + retrievedDataAmount++; + } finally { + retrievedDataAmountLock.unlock(); + } + } + + private void incrementFilteredDataAmount() { + filteredDataAmountLock.lock(); + try { + filteredDataAmount++; + } finally { + filteredDataAmountLock.unlock(); + } + } @Override protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java index e79eb57de25..25306052935 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java +++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java @@ -5,7 +5,8 @@ import java.util.HashSet; import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; -import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.logging.Level; import java.util.logging.Logger; @@ -143,29 +144,29 @@ public abstract class AbstractPartitioningParallelProcessor - extends AbstractPartitioningParallelProcessor, DataType, GroupedDataEntry> { + extends AbstractSimpleParallelProcessor> { private Iterable> dimensions; @@ -66,11 +66,6 @@ public class ParallelMultiDimensionalGroupingProcessor return new GenericGroupKey(keyValue); } - @Override - protected Iterable partitionElement(Iterable element) { - return element; - } - @Override protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelStoringAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelStoringAggregationProcessor.java index d383045679f..aaf91e7af4a 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelStoringAggregationProcessor.java +++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelStoringAggregationProcessor.java @@ -3,7 +3,8 @@ package com.sap.sse.datamining.impl.components.aggregators; import java.util.Collection; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; -import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import com.sap.sse.datamining.AdditionalResultDataBuilder; import com.sap.sse.datamining.components.Processor; @@ -12,13 +13,13 @@ import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor; public abstract class AbstractParallelStoringAggregationProcessor extends AbstractSimpleParallelProcessor { - private final ReentrantReadWriteLock storeLock; + private final Lock storeLock; private final String aggregationNameMessageKey; public AbstractParallelStoringAggregationProcessor(ExecutorService executor, Collection> resultReceivers, String aggregationNameMessageKey) { super(executor, resultReceivers); - storeLock = new ReentrantReadWriteLock(); + storeLock = new ReentrantLock(); this.aggregationNameMessageKey = aggregationNameMessageKey; } @@ -27,11 +28,11 @@ public abstract class AbstractParallelStoringAggregationProcessor() { @Override public AggregatedType call() throws Exception { - storeLock.writeLock().lock(); + storeLock.lock(); try { storeElement(element); } finally { - storeLock.writeLock().unlock(); + storeLock.unlock(); } return AbstractParallelStoringAggregationProcessor.super.createInvalidResult(); }