eclipse-sailing-analytics/sailing-analytics
1
0 Fork 0
mirror of https://github.com/eclipse-sailing-analytics/sailing-analytics.git synced 2026-11-08 03:52:18 +00:00
Code Activity

Merge branch 'master' into permission-vertical

Browse Source
This commit is contained in:
Jonas Dann committed 2017-11-28 18:41:19 +01:00
2 parents 9e2c6acd24 a505e4c69e
commit c9a617385d
1189 files changed
+26724 -8035
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/AbstractEnrichingProcessor.java
-1
View File
@@ -25,7 +25,6 @@ public abstract class AbstractEnrichingProcessor<InputType, ResultType> extends
@Override
protected ProcessorInstruction<ResultType> createInstruction(final InputType element) {
return new AbstractProcessorInstruction<ResultType>(this) {
@Override
public ResultType computeResult() {
return enrich(element);
java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/PolarDataMiner.java
+53 -49
View File
@@ -6,6 +6,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
@@ -13,9 +14,10 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;
import org.apache.commons.math.analysis.polynomials.PolynomialFunction;
@@ -42,6 +44,7 @@ import com.sap.sailing.domain.common.impl.PolarSheetsHistogramDataImpl;
import com.sap.sailing.domain.common.tracking.GPSFixMoving;
import com.sap.sailing.domain.polars.NotEnoughDataHasBeenAddedException;
import com.sap.sailing.domain.polars.PolarsChangedListener;
import com.sap.sailing.domain.tracking.GPSFixTrack;
import com.sap.sailing.domain.tracking.TrackedRace;
import com.sap.sailing.polars.impl.CubicEquation;
import com.sap.sse.datamining.components.FilterCriterion;
@@ -73,7 +76,10 @@ public class PolarDataMiner {
private static final int EXECUTOR_QUEUE_SIZE = 100;
private static final int THREAD_POOL_SIZE = ThreadPoolUtil.INSTANCE.getReasonableThreadPoolSize();
private final ThreadPoolExecutor executor = createExecutor();
private final ConcurrentMap<TrackedRace, Set<GPSFixMovingWithOriginInfo>> fixesForRacesWhichAreStillLoading = new ConcurrentHashMap<>();
private final ScheduledExecutorService processRacesThatFinishedLoadingExecutor = ThreadPoolUtil.INSTANCE
.createBackgroundTaskThreadPoolExecutor(1,
PolarDataMiner.class.getName() + " processRacesThatFinishedLoadingExecutor");
private final Map<BoatClass, AtomicInteger> stats = new ConcurrentHashMap<>();
private final Queue<GPSFixMovingWithOriginInfo> fixQueue = new ConcurrentLinkedQueue<GPSFixMovingWithOriginInfo>();
@@ -97,21 +103,13 @@ public class PolarDataMiner {
private ThreadPoolExecutor createExecutor() {
return new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE, 60l, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(EXECUTOR_QUEUE_SIZE), new ThreadFactoryWithPriority(PolarDataMiner.class.getSimpleName(),
Thread.NORM_PRIORITY-1, /* daemon */true), new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
logger.warning("Polar Data Miner Executor rejected execution. Running sequentially.");
r.run();
}
}) {
Thread.NORM_PRIORITY-1, /* daemon */true)) {
@Override
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
synchronized (fixQueue) {
while (this.getQueue().size() < (EXECUTOR_QUEUE_SIZE / 10) && !fixQueue.isEmpty()) {
GPSFixMovingWithOriginInfo fix = fixQueue.poll();
preFilteringProcessor.processElement(fix);
}
GPSFixMovingWithOriginInfo fix;
while (this.getQueue().size() < (EXECUTOR_QUEUE_SIZE / 10) && (fix=fixQueue.poll()) != null) {
preFilteringProcessor.processElement(fix);
}
}
};
@@ -193,7 +191,6 @@ public class PolarDataMiner {
preFilteringProcessor = new ParallelFilteringProcessor<GPSFixMovingWithOriginInfo>(
GPSFixMovingWithOriginInfo.class, executor, preFilterResultReceivers,
new FilterCriterion<GPSFixMovingWithOriginInfo>() {
@Override
public boolean matches(GPSFixMovingWithOriginInfo element) {
boolean result = false;
@@ -201,6 +198,14 @@ public class PolarDataMiner {
element.getCompetitor(),
backendPolarSheetGenerationSettings.getPctOfLeadingCompetitorsToInclude())) {
result = true;
final BoatClass boatClass = element.getCompetitor().getBoat().getBoatClass();
AtomicInteger count = stats.get(boatClass);
if (count == null) {
count = new AtomicInteger(1);
stats.put(boatClass, count);
} else {
count.getAndIncrement();
}
}
return result;
}
@@ -214,36 +219,20 @@ public class PolarDataMiner {
}
public void addFix(GPSFixMoving fix, Competitor competitor, TrackedRace trackedRace) {
GPSFixMovingWithOriginInfo fixWithOriginInfo = new GPSFixMovingWithOriginInfo(fix, trackedRace, competitor);
if (trackedRace.getStatus().getStatus() == TrackedRaceStatusEnum.LOADING) {
/*
* logger.info("Received fix for replay race which has not finished loading. Queuing. " +
* (trackedRace.getRace() != null ? trackedRace.getRace().getName() : trackedRace.getRaceIdentifier()
* .getRaceName()));
*/
Set<GPSFixMovingWithOriginInfo> fixes = fixesForRacesWhichAreStillLoading.get(trackedRace);
if (fixes == null) {
synchronized (fixesForRacesWhichAreStillLoading) {
fixes = fixesForRacesWhichAreStillLoading.get(trackedRace);
if (fixes == null) {
fixes = new HashSet<>();
fixesForRacesWhichAreStillLoading.put(trackedRace, fixes);
}
}
}
fixes.add(fixWithOriginInfo);
} else {
// don't process fixes while LOADING because wind data is loading at the same time, and
// unpredictable results may occur due to this
if (trackedRace.getStatus().getStatus() != TrackedRaceStatusEnum.LOADING) {
GPSFixMovingWithOriginInfo fixWithOriginInfo = new GPSFixMovingWithOriginInfo(fix, trackedRace, competitor);
processFix(trackedRace, fixWithOriginInfo);
}
}
private void processFix(TrackedRace trackedRace, GPSFixMovingWithOriginInfo fixWithOriginInfo) {
synchronized (fixQueue) {
if (executor.getQueue().size() >= EXECUTOR_QUEUE_SIZE / 10) {
fixQueue.add(fixWithOriginInfo);
} else {
preFilteringProcessor.processElement(fixWithOriginInfo);
}
if (executor.getQueue().size() >= EXECUTOR_QUEUE_SIZE / 10) { // in this case synchronous execution becomes likely, but
// we are in a synchronous callback and don't want to spend too much time here in this foreground thread; queue it!
fixQueue.add(fixWithOriginInfo);
} else {
preFilteringProcessor.processElement(fixWithOriginInfo);
}
}
@@ -433,18 +422,33 @@ public class PolarDataMiner {
return speedRegressionPerAngleClusterProcessor.getSpeedRegressionFunction(boatClass, trueWindAngle);
}
public void raceFinishedTracking(TrackedRace race) {
Set<GPSFixMovingWithOriginInfo> fixes = null;
synchronized (fixesForRacesWhichAreStillLoading) {
fixes = fixesForRacesWhichAreStillLoading.remove(race);
}
if (fixes != null) {
public void raceFinishedTracking(final TrackedRace race) {
processRacesThatFinishedLoadingExecutor.execute(()->{
logger.info("All queued fixes for newly loaded race will process now. "
+ (race.getRace() != null ? race.getRace().getName() : race.getRaceIdentifier().getRaceName()));
for (GPSFixMovingWithOriginInfo fix : fixes) {
processFix(race, fix);
for (final Competitor competitor : race.getRace().getCompetitors()) {
final GPSFixTrack<Competitor, GPSFixMoving> track = race.getTrack(competitor);
// it is necessary to release the track's lock before calling processElement
// because processElement will transitively cause obtaining the course lock,
// and other methods will first obtain the course and then the track lock, leading
// to a deadlock. See also bug 4297.
final List<GPSFixMoving> fixes = new ArrayList<>();
track.lockForRead();
try {
for (final GPSFixMoving fix : track.getFixes()) {
fixes.add(fix);
}
} finally {
track.unlockAfterRead();
}
for (final GPSFixMoving fix : fixes) {
preFilteringProcessor.processElement(new GPSFixMovingWithOriginInfo(fix, race, competitor));
}
}
}
logger.info("Finished injecting fixes for race "
+ (race.getRace() != null ? race.getRace().getName() : race.getRaceIdentifier().getRaceName())
+ "; stats: " + stats);
});
}
public void registerListener(BoatClass boatClass, PolarsChangedListener listener) {