From 05e4c0cf5c253bac11d5dbbaa31c778b1f929326 Mon Sep 17 00:00:00 2001 From: Axel Uhl Date: Fri, 18 Aug 2017 17:00:11 +0200 Subject: [PATCH] use a single background thread to process polar sheet to avoid queuing Change-Id: I3c88c7e9ca5f734327a0a75e35b19cc38f4b3ef2 --- .../mining/test/PolarDataMinerTest.java | 8 ++- .../sailing/polars/mining/PolarDataMiner.java | 67 ++++++++++--------- .../components/AbstractParallelProcessor.java | 2 +- .../src/com/sap/sse/util/ThreadPoolUtil.java | 2 + .../sap/sse/util/impl/ThreadPoolUtilImpl.java | 5 ++ 5 files changed, 51 insertions(+), 33 deletions(-) diff --git a/java/com.sap.sailing.polars.test/src/com/sap/sailing/polars/mining/test/PolarDataMinerTest.java b/java/com.sap.sailing.polars.test/src/com/sap/sailing/polars/mining/test/PolarDataMinerTest.java index 54b47b9586b..cf39e57c376 100644 --- a/java/com.sap.sailing.polars.test/src/com/sap/sailing/polars/mining/test/PolarDataMinerTest.java +++ b/java/com.sap.sailing.polars.test/src/com/sap/sailing/polars/mining/test/PolarDataMinerTest.java @@ -33,6 +33,7 @@ import com.sap.sailing.domain.common.Bearing; import com.sap.sailing.domain.common.PolarSheetGenerationSettings; import com.sap.sailing.domain.common.Position; import com.sap.sailing.domain.common.SpeedWithBearing; +import com.sap.sailing.domain.common.TrackedRaceStatusEnum; import com.sap.sailing.domain.common.Wind; import com.sap.sailing.domain.common.impl.DegreeBearingImpl; import com.sap.sailing.domain.common.impl.DegreePosition; @@ -46,6 +47,7 @@ import com.sap.sailing.domain.polars.NotEnoughDataHasBeenAddedException; import com.sap.sailing.domain.tracking.DynamicGPSFixTrack; import com.sap.sailing.domain.tracking.MarkPassing; import com.sap.sailing.domain.tracking.TrackedRace; +import com.sap.sailing.domain.tracking.TrackedRaceStatus; import com.sap.sailing.domain.tracking.WindWithConfidence; import com.sap.sailing.domain.tracking.impl.DynamicGPSFixMovingTrackImpl; import com.sap.sailing.domain.tracking.impl.WindWithConfidenceImpl; @@ -82,7 +84,7 @@ public class PolarDataMinerTest { return new BearingClusterGroup(0, 180, 5); } - @Ignore + @Ignore("The test did work before DataMining was used for polars; maybe rework in the future...") @Test public void testGrouping() throws InterruptedException, TimeoutException, NoSuchMethodException, NotEnoughDataHasBeenAddedException { @@ -203,7 +205,9 @@ public class PolarDataMinerTest { MarkPassing markpassing = createMockedStartMarkPassing(); when(trackedRace.getMarkPassing(eq(competitor), eq(startWaypoint))).thenReturn(markpassing); } - + TrackedRaceStatus status = mock(TrackedRaceStatus.class); + when(status.getStatus()).thenReturn(TrackedRaceStatusEnum.FINISHED); + when(trackedRace.getStatus()).thenReturn(status); when(trackedRace.getStartOfRace()).thenReturn(startOfRace); diff --git a/java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/PolarDataMiner.java b/java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/PolarDataMiner.java index c28abadf223..36f870a3ed9 100755 --- a/java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/PolarDataMiner.java +++ b/java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/PolarDataMiner.java @@ -13,8 +13,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.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; @@ -41,6 +43,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; @@ -72,7 +75,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> fixesForRacesWhichAreStillLoading = new ConcurrentHashMap<>(); + private final ScheduledExecutorService processRacesThatFinishedLoadingExecutor = ThreadPoolUtil.INSTANCE + .createBackgroundTaskThreadPoolExecutor(1, + PolarDataMiner.class.getName() + " processRacesThatFinishedLoadingExecutor"); + private final Map stats = new ConcurrentHashMap<>(); private final Queue fixQueue = new ConcurrentLinkedQueue(); @@ -191,6 +197,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; } @@ -204,31 +218,17 @@ 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 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) { - if (executor.getQueue().size() >= EXECUTOR_QUEUE_SIZE / 10) { + 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); @@ -421,18 +421,25 @@ public class PolarDataMiner { return speedRegressionPerAngleClusterProcessor.getSpeedRegressionFunction(boatClass, trueWindAngle); } - public void raceFinishedTracking(TrackedRace race) { - final Set fixes; - 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 track = race.getTrack(competitor); + track.lockForRead(); + try { + for (final GPSFixMoving fix : track.getFixes()) { + preFilteringProcessor.processElement(new GPSFixMovingWithOriginInfo(fix, race, competitor)); + } + } finally { + track.unlockAfterRead(); + } } - } + 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) { diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java index 8098ad85580..caa9f427424 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java +++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java @@ -48,7 +48,7 @@ public abstract class AbstractParallelProcessor extends A try { executor.execute(instruction); } catch (RejectedExecutionException exc) { - LOGGER.log(Level.WARNING, "A " + RejectedExecutionException.class.getSimpleName() + LOGGER.log(Level.FINEST, "A " + RejectedExecutionException.class.getSimpleName() + " appeared during the processing."); instruction.run(); } diff --git a/java/com.sap.sse/src/com/sap/sse/util/ThreadPoolUtil.java b/java/com.sap.sse/src/com/sap/sse/util/ThreadPoolUtil.java index 16fb3576b42..d257f585c2c 100644 --- a/java/com.sap.sse/src/com/sap/sse/util/ThreadPoolUtil.java +++ b/java/com.sap.sse/src/com/sap/sse/util/ThreadPoolUtil.java @@ -54,6 +54,8 @@ public interface ThreadPoolUtil { */ ScheduledExecutorService createBackgroundTaskThreadPoolExecutor(String name); + ScheduledExecutorService createBackgroundTaskThreadPoolExecutor(int size, String name); + /** * In case an application module really requires its own thread pool instead of * using {@link #getBackgroundTaskThreadPool}, it should size the thread pool diff --git a/java/com.sap.sse/src/com/sap/sse/util/impl/ThreadPoolUtilImpl.java b/java/com.sap.sse/src/com/sap/sse/util/impl/ThreadPoolUtilImpl.java index f14f9445678..4c6b046a7a3 100644 --- a/java/com.sap.sse/src/com/sap/sse/util/impl/ThreadPoolUtilImpl.java +++ b/java/com.sap.sse/src/com/sap/sse/util/impl/ThreadPoolUtilImpl.java @@ -30,6 +30,11 @@ public class ThreadPoolUtilImpl implements ThreadPoolUtil { return createThreadPoolExecutor(name, Thread.NORM_PRIORITY-1); } + @Override + public ScheduledExecutorService createBackgroundTaskThreadPoolExecutor(int size, String name) { + return createThreadPoolExecutor(name, Thread.NORM_PRIORITY-1, size); + } + @Override public ScheduledExecutorService createForegroundTaskThreadPoolExecutor(String name) { return createThreadPoolExecutor(name, Thread.NORM_PRIORITY);