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

Merge branch 'bug3928'

Browse Source
This commit is contained in:
Axel Uhl committed 2016-09-28 17:59:54 +02:00
2 parents 0ab64a06e3 ed7385a487
commit c01daef9b9
30 files changed
+698 -172
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

java/com.sap.sailing.domain.shared.android/src/com/sap/sailing/domain/base/impl/WaypointImpl.java
+1 -1
View File
@@ -46,7 +46,7 @@ public class WaypointImpl implements Waypoint {
@Override
public String toString() {
return getName();
return (getName()==null?"":getName()) + " ("+getPassingInstructions()+")";
}
@Override
java/com.sap.sailing.domain.test/src/com/sap/sailing/domain/test/ConcurrencyTest.java
+2 -2
View File
@@ -10,7 +10,6 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.text.ParseException;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.logging.Logger;
import org.junit.Before;
@@ -30,6 +29,7 @@ import com.sap.sailing.domain.tracking.MarkPassing;
import com.sap.sailing.domain.tractracadapter.ReceiverType;
import com.sap.sailing.domain.tractracadapter.TracTracConnectionConstants;
import com.sap.sse.common.impl.MillisecondsTimePoint;
import com.sap.sse.util.ThreadPoolUtil;
import com.tractrac.model.lib.api.event.CreateModelException;
import com.tractrac.subscription.lib.api.SubscriberInitializationException;
@@ -77,7 +77,7 @@ public class ConcurrencyTest extends OnlineTracTracBasedTest {
logger.info("1 thread: "+duration+"ns");
logger.info("number of approximation points: "+approximation1.size());
dp = new DouglasPeucker<Competitor, GPSFixMoving>(teamAquaTrack, Executors.newFixedThreadPool(numberOfCPUs));
dp = new DouglasPeucker<Competitor, GPSFixMoving>(teamAquaTrack, ThreadPoolUtil.INSTANCE.getDefaultBackgroundTaskThreadPoolExecutor());
start = System.nanoTime();
List<GPSFixMoving> approximation2 = dp.approximate(new MeterDistance(3), firstMarkPassing.getTimePoint(), lastMarkPassing.getTimePoint());
duration = System.nanoTime()-start;
java/com.sap.sailing.domain/src/com/sap/sailing/domain/base/impl/AbstractRaceExecutionOrderProvider.java
+8 -2
View File
@@ -43,6 +43,10 @@ public abstract class AbstractRaceExecutionOrderProvider implements RaceExecutio
protected abstract Map<Fleet, Iterable<? extends RaceColumn>> getRaceColumnsOfSeries();
/**
* For all tracked races found on any of the {@link #getRaceColumnsOfSeries() race columns} produces a valid and
* potentially empty set.
*/
private Map<TrackedRace, Set<TrackedRace>> reloadAndGetPreviousRacesByRace() {
final Map<TrackedRace, Set<TrackedRace>> previousRacesByRace = new HashMap<>();
for (Entry<Fleet, Iterable<? extends RaceColumn>> raceColumnsInSeries : getRaceColumnsOfSeries().entrySet()) {
@@ -96,11 +100,13 @@ public abstract class AbstractRaceExecutionOrderProvider implements RaceExecutio
@Override
public Set<TrackedRace> getPreviousRacesInExecutionOrder(TrackedRace race) {
Set<TrackedRace> result = Collections.emptySet();
final Set<TrackedRace> result;
final Map<TrackedRace, Set<TrackedRace>> previousRacesByRace = getPreviousRacesByRace();
if (previousRacesByRace != null) {
result = previousRacesByRace.get(race);
}
} else {
result = Collections.emptySet();
}
return result;
}
}
java/com.sap.sailing.domain/src/com/sap/sailing/domain/base/impl/DouglasPeucker.java
+5 -15
View File
@@ -6,12 +6,9 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ScheduledExecutorService;
import com.sap.sailing.domain.common.Bearing;
import com.sap.sailing.domain.common.Distance;
@@ -31,7 +28,7 @@ import com.sap.sse.common.Util;
public class DouglasPeucker<ItemType, FixType extends GPSFix> {
private final GPSFixTrack<ItemType, FixType> track;
private final Executor executor;
private final ScheduledExecutorService executor;
public DouglasPeucker(GPSFixTrack<ItemType, FixType> track) {
this.track = track;
@@ -42,7 +39,7 @@ public class DouglasPeucker<ItemType, FixType extends GPSFix> {
* Allows clients to specify a non-default, optionally multi-threaded executor that will be used to perform the
* high-effort computations.
*/
public DouglasPeucker(GPSFixTrack<ItemType, FixType> track, Executor executor) {
public DouglasPeucker(GPSFixTrack<ItemType, FixType> track, ScheduledExecutorService executor) {
this.track = track;
this.executor = executor;
}
@@ -96,15 +93,8 @@ public class DouglasPeucker<ItemType, FixType extends GPSFix> {
if (fix.getTimePoint().compareTo(to) > 0) {
break;
}
RunnableFuture<Util.Pair<FixType, Distance>> crossTrackErrorFuture = new FutureTask<Util.Pair<FixType, Distance>>(
new Callable<Util.Pair<FixType, Distance>>() {
@Override
public Util.Pair<FixType, Distance> call() throws Exception {
return new Util.Pair<FixType, Distance>(fix, fix.getPosition().absoluteCrossTrackError(
fromPosition, bearing));
}
});
executor.execute(crossTrackErrorFuture);
Future<Util.Pair<FixType, Distance>> crossTrackErrorFuture = executor.submit(
()->new Util.Pair<FixType, Distance>(fix, fix.getPosition().absoluteCrossTrackError(fromPosition, bearing)));
crossTrackErrorFutures.add(crossTrackErrorFuture);
}
for (Future<Util.Pair<FixType, Distance>> crossTrackErrorFuture : crossTrackErrorFutures) {
java/com.sap.sailing.domain/src/com/sap/sailing/domain/leaderboard/caching/LeaderboardDTOCache.java
+11 -9
View File
@@ -23,6 +23,7 @@ import com.sap.sse.common.TimePoint;
import com.sap.sse.common.Util;
import com.sap.sse.concurrent.LockUtil;
import com.sap.sse.concurrent.NamedReentrantReadWriteLock;
import com.sap.sse.util.impl.FutureTaskWithTracingGet;
/**
* Caches the expensive to compute {@link LeaderboardDTO} results of a
@@ -140,15 +141,16 @@ public class LeaderboardDTOCache implements LeaderboardCache {
LockUtil.unlockAfterRead(leaderboardCacheLock);
}
if (future == null) {
future = new FutureTask<LeaderboardDTO>(new Callable<LeaderboardDTO>() {
@Override
public LeaderboardDTO call() throws Exception {
LeaderboardDTO result = leaderboard.computeDTO(adjustedTimePoint,
namesOfRaceColumnsForWhichToLoadLegDetails, addOverallDetails,
waitForLatestAnalyses, trackedRegattaRegistry, baseDomainFactory, /* fillTotalPointsUncorrected */ false);
return result;
}
});
future = new FutureTaskWithTracingGet<LeaderboardDTO>(LeaderboardDTOCache.class.getName()+" for leaderboard "+leaderboard.getName(),
new Callable<LeaderboardDTO>() {
@Override
public LeaderboardDTO call() throws Exception {
LeaderboardDTO result = leaderboard.computeDTO(adjustedTimePoint,
namesOfRaceColumnsForWhichToLoadLegDetails, addOverallDetails,
waitForLatestAnalyses, trackedRegattaRegistry, baseDomainFactory, /* fillTotalPointsUncorrected */ false);
return result;
}
});
// The add(Leaderboard) method that the cache manager calls back on this class does nothing, so no synchronization required
this.leaderboardCacheManager.add(leaderboard); // ensure the leaderboard is tracked for changes to invalidate
LockUtil.lockForWrite(leaderboardCacheLock);
java/com.sap.sailing.domain/src/com/sap/sailing/domain/leaderboard/impl/AbstractSimpleLeaderboardImpl.java
+8 -10
View File
@@ -21,7 +21,6 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RunnableFuture;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -97,6 +96,7 @@ import com.sap.sse.common.impl.MillisecondsTimePoint;
import com.sap.sse.concurrent.LockUtil;
import com.sap.sse.concurrent.NamedReentrantReadWriteLock;
import com.sap.sse.util.ThreadPoolUtil;
import com.sap.sse.util.impl.FutureTaskWithTracingGet;
/**
* Base implementation for various types of leaderboards. The {@link RaceColumnListener} implementation forwards events
@@ -311,7 +311,7 @@ public abstract class AbstractSimpleLeaderboardImpl implements Leaderboard, Race
this.suppressedCompetitors = new HashSet<Competitor>();
this.suppressedCompetitorsLock = new NamedReentrantReadWriteLock("suppressedCompetitorsLock", /* fair */ false);
this.raceColumnListeners = new RaceColumnListeners();
this.raceDetailsAtEndOfTrackingCache = new HashMap<com.sap.sse.common.Util.Pair<TrackedRace, Competitor>, RunnableFuture<RaceDetails>>();
this.raceDetailsAtEndOfTrackingCache = new HashMap<>();
initTransientFields();
}
@@ -330,7 +330,7 @@ public abstract class AbstractSimpleLeaderboardImpl implements Leaderboard, Race
}
private void initTransientFields() {
this.raceDetailsAtEndOfTrackingCache = new HashMap<com.sap.sse.common.Util.Pair<TrackedRace,Competitor>, RunnableFuture<RaceDetails>>();
this.raceDetailsAtEndOfTrackingCache = new HashMap<>();
this.cacheInvalidationListeners = new HashSet<CacheInvalidationListener>();
// When many updates are triggered in a short period of time by a single thread, ensure that the single thread
// providing the updates is not outperformed by all the re-calculations happening here. Leave at least one
@@ -1225,7 +1225,7 @@ public abstract class AbstractSimpleLeaderboardImpl implements Leaderboard, Race
}
// Now create the race columns and, as a future task, set their competitorsFromBestToWorst, then wait for all these
// futures to finish:
Map<RaceColumn, FutureTask<List<CompetitorDTO>>> competitorsFromBestToWorstTasks = new HashMap<>();
Map<RaceColumn, Future<List<CompetitorDTO>>> competitorsFromBestToWorstTasks = new HashMap<>();
for (final RaceColumn raceColumn : this.getRaceColumns()) {
boolean isMetaLeaderboardColumn = raceColumn instanceof MetaLeaderboardColumn;
RaceColumnDTO raceColumnDTO = result.createEmptyRaceColumn(raceColumn.getName(), raceColumn.isMedalRace(),
@@ -1253,13 +1253,12 @@ public abstract class AbstractSimpleLeaderboardImpl implements Leaderboard, Race
raceColumn instanceof RaceColumnInSeries ? ((RaceColumnInSeries) raceColumn).getSeries().getName() : null,
fleetDTO, raceColumn.isMedalRace(), raceIdentifier, race, isMetaLeaderboardColumn);
}
FutureTask<List<CompetitorDTO>> task = new FutureTask<List<CompetitorDTO>>(
Future<List<CompetitorDTO>> task = executor.submit(
() -> baseDomainFactory.getCompetitorDTOList(AbstractSimpleLeaderboardImpl.this.getCompetitorsFromBestToWorst(raceColumn, timePoint)));
executor.execute(task);
competitorsFromBestToWorstTasks.put(raceColumn, task);
}
// wait for the competitor orderings to have been computed for all race columns before continuing; subsequent tasks may depend on these data
for (Map.Entry<RaceColumn, FutureTask<List<CompetitorDTO>>> raceColumnAndTaskToJoin : competitorsFromBestToWorstTasks.entrySet()) {
for (Map.Entry<RaceColumn, Future<List<CompetitorDTO>>> raceColumnAndTaskToJoin : competitorsFromBestToWorstTasks.entrySet()) {
try {
result.setCompetitorsFromBestToWorst(raceColumnAndTaskToJoin.getKey().getName(), raceColumnAndTaskToJoin.getValue().get());
} catch (InterruptedException e) {
@@ -1323,7 +1322,7 @@ public abstract class AbstractSimpleLeaderboardImpl implements Leaderboard, Race
Map<String, Future<LeaderboardEntryDTO>> futuresForColumnName = new HashMap<String, Future<LeaderboardEntryDTO>>();
final Set<RaceColumn> discardedRaceColumns = getResultDiscardingRule().getDiscardedRaceColumns(competitor, this, getRaceColumns(), timePoint);
for (final RaceColumn raceColumn : this.getRaceColumns()) {
RunnableFuture<LeaderboardEntryDTO> future = new FutureTask<LeaderboardEntryDTO>(() -> {
Future<LeaderboardEntryDTO> future = executor.submit(() -> {
Entry entry = AbstractSimpleLeaderboardImpl.this.getEntry(competitor, raceColumn, timePoint, discardedRaceColumns);
return getLeaderboardEntryDTO(entry, raceColumn, competitor, timePoint,
namesOfRaceColumnsForWhichToLoadLegDetails != null
@@ -1331,7 +1330,6 @@ public abstract class AbstractSimpleLeaderboardImpl implements Leaderboard, Race
.getName()), waitForLatestAnalyses, legRanksCache, baseDomainFactory,
fillTotalPointsUncorrected, cache);
});
executor.execute(future);
futuresForColumnName.put(raceColumn.getName(), future);
}
for (Map.Entry<String, Future<LeaderboardEntryDTO>> raceColumnNameAndFuture : futuresForColumnName.entrySet()) {
@@ -1611,7 +1609,7 @@ public abstract class AbstractSimpleLeaderboardImpl implements Leaderboard, Race
raceDetails = raceDetailsAtEndOfTrackingCache.get(key);
if (raceDetails == null) {
needToRunRaceDetails = true;
raceDetails = new FutureTask<RaceDetails>(new Callable<RaceDetails>() {
raceDetails = new FutureTaskWithTracingGet<RaceDetails>("RaceDetails for "+trackedRace, new Callable<RaceDetails>() {
@Override
public RaceDetails call() throws Exception {
TimePoint end = trackedRace.getEndOfRace();
java/com.sap.sailing.domain/src/com/sap/sailing/domain/markpassingcalculation/MarkPassingCalculator.java
+38 -15
View File
@@ -9,7 +9,6 @@ import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -173,7 +172,7 @@ public class MarkPassingCalculator {
+ " while waiting for new GPSFixes");
}
listener.getQueue().drainTo(allNewFixInsertions);
logger.finer("MPC for "+raceName+" recieved "+ allNewFixInsertions.size()+" new updates.");
logger.fine("MPC for "+raceName+" received "+ allNewFixInsertions.size()+" new updates.");
for (StorePositionUpdateStrategy fixInsertion : allNewFixInsertions) {
if (listener.isEndMarker(fixInsertion)) {
logger.info("Stopping "+MarkPassingCalculator.this+"'s listener for race "+raceName);
@@ -212,8 +211,7 @@ public class MarkPassingCalculator {
}
executor.invokeAll(tasks);
}
updateManuallySetMarkPassings(fixedMarkPassings, removedFixedMarkPassings, suppressedMarkPassings,
unsuppressedMarkPassings);
updateManuallySetMarkPassings(fixedMarkPassings, removedFixedMarkPassings, suppressedMarkPassings, unsuppressedMarkPassings);
computeMarkPasses(competitorFixes, markFixes);
competitorFixes.clear();
markFixes.clear();
@@ -284,9 +282,24 @@ public class MarkPassingCalculator {
fixes.addAll(fixesAffectedByNewMarkFixes.getValue());
}
}
List<Callable<Object>> tasks = new ArrayList<>();
for (Entry<Competitor, Set<GPSFix>> c : combinedCompetitorFixes.entrySet()) {
tasks.add(Executors.callable(new ComputeMarkPassings(c.getKey(), c.getValue())));
List<Callable<Void>> tasks = new ArrayList<>();
for (final Entry<Competitor, Set<GPSFix>> c : combinedCompetitorFixes.entrySet()) {
final Runnable runnable = new ComputeMarkPassings(c.getKey(), c.getValue());
tasks.add(new Callable<Void>() {
@Override
public Void call() throws Exception {
runnable.run();
return null;
}
/**
* A reasonable toString implementation helps identifying these tasks in log messages for long-running tasks
*/
@Override
public String toString() {
return "Mark passing calculation for competitor "+c.getKey()+" with "+c.getValue().size()+" fixes";
}
});
}
try {
executor.invokeAll(tasks);
@@ -296,8 +309,8 @@ public class MarkPassingCalculator {
}
private class ComputeMarkPassings implements Runnable {
Competitor c;
Iterable<GPSFix> fixes;
final Competitor c;
final Iterable<GPSFix> fixes;
public ComputeMarkPassings(Competitor c, Iterable<GPSFix> fixes) {
this.c = c;
@@ -306,12 +319,22 @@ public class MarkPassingCalculator {
@Override
public void run() {
logger.finer("Calculating MarkPassings for race "+raceName+", competitor " + c + " (" + Util.size(fixes) + " new fixes)");
Util.Pair<Iterable<Candidate>, Iterable<Candidate>> candidateDeltas = finder.getCandidateDeltas(c, fixes);
logger.finer("Received " + Util.size(candidateDeltas.getA()) + " new Candidates for race "+raceName+
" and competitor "+c+" and will remove "
+ Util.size(candidateDeltas.getB()) + " old Candidates for " + c);
chooser.calculateMarkPassDeltas(c, candidateDeltas.getA(), candidateDeltas.getB());
try {
if (logger.isLoggable(Level.FINER)) {
logger.finer("Calculating MarkPassings for race "+raceName+", competitor " + c + " (" + Util.size(fixes) + " new fixes)");
}
Util.Pair<Iterable<Candidate>, Iterable<Candidate>> candidateDeltas = finder.getCandidateDeltas(c, fixes);
if (logger.isLoggable(Level.FINER)) {
logger.finer("Received " + Util.size(candidateDeltas.getA()) + " new Candidates for race "+raceName+
" and competitor "+c+" and will remove "
+ Util.size(candidateDeltas.getB()) + " old Candidates for " + c);
}
chooser.calculateMarkPassDeltas(c, candidateDeltas.getA(), candidateDeltas.getB());
} catch (Exception e) {
// make sure the exception is logged and not only "swallowed" and suppressed by an invokeAll(...) statement
logger.log(Level.SEVERE, "Exception trying to compute mark passings for competitor "+c, e);
throw e;
}
}
}
java/com.sap.sailing.domain/src/com/sap/sailing/domain/markpassingcalculation/impl/CandidateFinderImpl.java
+35 -23
View File
@@ -47,6 +47,7 @@ import com.sap.sse.common.Timed;
import com.sap.sse.common.Util;
import com.sap.sse.common.Util.Pair;
import com.sap.sse.common.impl.TimeRangeImpl;
import com.sap.sse.concurrent.LockUtil;
import com.sap.sse.util.ThreadPoolUtil;
/**
@@ -383,24 +384,30 @@ public class CandidateFinderImpl implements CandidateFinder {
changedWaypoints.add(w);
}
}
final Set<Callable<Void>> tasks = new HashSet<>();;
final Set<Callable<Void>> tasks = new HashSet<>();
final Thread executingThread = Thread.currentThread(); // most likely the MarkPassingCalculator.Listen thread
for (Competitor c : race.getRace().getCompetitors()) {
tasks.add(()->{
List<Candidate> badCans = new ArrayList<>();
List<Candidate> newCans = new ArrayList<>();
for (Waypoint w : changedWaypoints) {
Map<List<GPSFix>, Candidate> xteCans = getXteCandidates(c, w);
badCans.addAll(xteCans.values());
xteCans.clear();
Map<GPSFix, Candidate> distanceCans = getDistanceCandidates(c, w);
badCans.addAll(distanceCans.values());
distanceCans.clear();
LockUtil.propagateLockSetFrom(executingThread); // "inherit" the course read lock from the "calling" thread into the thread pool executor thread
try {
List<Candidate> badCans = new ArrayList<>();
List<Candidate> newCans = new ArrayList<>();
for (Waypoint w : changedWaypoints) {
Map<List<GPSFix>, Candidate> xteCans = getXteCandidates(c, w);
badCans.addAll(xteCans.values());
xteCans.clear();
Map<GPSFix, Candidate> distanceCans = getDistanceCandidates(c, w);
badCans.addAll(distanceCans.values());
distanceCans.clear();
}
Set<GPSFix> allFixes = getAllFixes(c);
newCans.addAll(checkForDistanceCandidateChanges(c, allFixes, changedWaypoints).getA());
newCans.addAll(checkForXTECandidatesChanges(c, allFixes, changedWaypoints).getA());
result.put(c, new Util.Pair<List<Candidate>, List<Candidate>>(newCans, badCans));
return null;
} finally {
LockUtil.unpropagateLockSetFrom(executingThread); // "return" the course read lock from the thread pool executor thread to the "calling" thread
}
Set<GPSFix> allFixes = getAllFixes(c);
newCans.addAll(checkForDistanceCandidateChanges(c, allFixes, changedWaypoints).getA());
newCans.addAll(checkForXTECandidatesChanges(c, allFixes, changedWaypoints).getA());
result.put(c, new Util.Pair<List<Candidate>, List<Candidate>>(newCans, badCans));
return null;
});
}
executor.invokeAll(tasks);
@@ -513,8 +520,8 @@ public class CandidateFinderImpl implements CandidateFinder {
private Set<GPSFix> getAllFixes(Competitor c) {
Set<GPSFix> fixes = new TreeSet<GPSFix>(comp);
DynamicGPSFixTrack<Competitor, GPSFixMoving> track = race.getTrack(c);
track.lockForRead();
try {
track.lockForRead();
for (GPSFix fix : track.getFixes(
timeRangeForValidCandidates.from(), /* fromInclusive */ true,
timeRangeForValidCandidates.to(), /* toInclusive */ true)) {
@@ -542,8 +549,8 @@ public class CandidateFinderImpl implements CandidateFinder {
affectedFixes.add(fix);
GPSFix fixBefore;
GPSFix fixAfter;
track.lockForRead();
try {
track.lockForRead();
TimePoint timePoint = fix.getTimePoint();
fixBefore = track.getLastFixBefore(timePoint);
fixAfter = track.getFirstFixAfter(timePoint);
@@ -1206,19 +1213,24 @@ public class CandidateFinderImpl implements CandidateFinder {
* @return an average of the estimated legs before and after <code>w</code>.
*/
private Distance getAverageLengthOfAdjacentLegs(TimePoint t, Waypoint w) {
Course course = race.getRace().getCourse();
if (w == course.getFirstWaypoint()) {
return race.getTrackedLegStartingAt(w).getGreatCircleDistance(t);
final Distance result;
final Course course = race.getRace().getCourse();
if (course.getNumberOfWaypoints() < 2) {
result = null;
} else if (w == course.getFirstWaypoint()) {
result = race.getTrackedLegStartingAt(w).getGreatCircleDistance(t);
} else if (w == course.getLastWaypoint()) {
return race.getTrackedLegFinishingAt(w).getGreatCircleDistance(t);
result = race.getTrackedLegFinishingAt(w).getGreatCircleDistance(t);
} else {
Distance before = race.getTrackedLegStartingAt(w).getGreatCircleDistance(t);
Distance after = race.getTrackedLegFinishingAt(w).getGreatCircleDistance(t);
if (after != null && before != null) {
return new MeterDistance(before.add(after).getMeters() / 2);
result = new MeterDistance(before.add(after).getMeters() / 2);
} else {
result = null;
}
return null;
}
return result;
}
/**
java/com.sap.sailing.domain/src/com/sap/sailing/domain/tracking/impl/TrackedRaceImpl.java
+3 -2
View File
@@ -158,6 +158,7 @@ import com.sap.sse.util.SmartFutureCache;
import com.sap.sse.util.SmartFutureCache.AbstractCacheUpdater;
import com.sap.sse.util.SmartFutureCache.EmptyUpdateInterval;
import com.sap.sse.util.impl.ArrayListNavigableSet;
import com.sap.sse.util.impl.FutureTaskWithTracingGet;
import difflib.DiffUtils;
import difflib.Patch;
@@ -750,7 +751,7 @@ public abstract class TrackedRaceImpl extends TrackedRaceWithWindEssentials impl
/**
* Monitor object to synchronize access to the {@link #updateStartAndEndOfTracking(boolean)} method. See bug 3922.
*/
private final Serializable updateStartAndEndOfTrackingMonitor = "";
private final Serializable updateStartAndEndOfTrackingMonitor = "updateStartAndEndOfTrackingMonitor";
/**
* Updates the start and end of tracking in the following precedence order:
@@ -1875,7 +1876,7 @@ public abstract class TrackedRaceImpl extends TrackedRaceWithWindEssentials impl
synchronized (directionFromStartToNextMarkCache) {
future = directionFromStartToNextMarkCache.get(at);
if (future == null) {
newFuture = new FutureTask<Wind>(new Callable<Wind>() {
newFuture = new FutureTaskWithTracingGet<Wind>("getDirectionFromStartToNextMark for "+this, new Callable<Wind>() {
@Override
public Wind call() {
Wind result;
java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/adminconsole/SimulationPanel.java
+60
View File
@@ -0,0 +1,60 @@
package com.sap.sailing.gwt.ui.adminconsole;
import com.google.gwt.event.dom.client.ClickEvent;
import com.google.gwt.event.dom.client.ClickHandler;
import com.google.gwt.user.client.ui.CheckBox;
import com.google.gwt.user.client.ui.FlowPanel;
import com.google.gwt.user.client.ui.HorizontalPanel;
import com.google.gwt.user.client.ui.Label;
import com.google.gwt.user.client.ui.TextBox;
import com.google.gwt.user.client.ui.VerticalPanel;
import com.sap.sailing.gwt.ui.client.StringMessages;
/**
* Can be used in {@link AbstractEventManagementPanel}s to enable some sort of time-delayed race simulation.
* Provides a checkbox (see {@link #isSimulate()}) that enables simulation and an offset in minutes (see
* {@link #getOffsetToStartTimeInMinutes()}).
*
* @author Axel Uhl (d043530)
*
*/
public class SimulationPanel extends VerticalPanel {
private static final String ZERO_AS_STRING = "0";
final CheckBox simulateWithStartTimeNowCheckBox;
final TextBox offsetToStartTimeOfSimulatedRaceTextBox;
public SimulationPanel(StringMessages stringMessages) {
offsetToStartTimeOfSimulatedRaceTextBox = new TextBox();
offsetToStartTimeOfSimulatedRaceTextBox.setWidth("40px");
offsetToStartTimeOfSimulatedRaceTextBox.setEnabled(false);
offsetToStartTimeOfSimulatedRaceTextBox.setValue(ZERO_AS_STRING);
simulateWithStartTimeNowCheckBox = new CheckBox(stringMessages.simulateAsLiveRace());
simulateWithStartTimeNowCheckBox.ensureDebugId("SimulateWithStartTimeNowCheckBox");
simulateWithStartTimeNowCheckBox.setWordWrap(false);
simulateWithStartTimeNowCheckBox.setValue(Boolean.FALSE);
simulateWithStartTimeNowCheckBox.addClickHandler(new ClickHandler() {
@Override
public void onClick(ClickEvent event) {
offsetToStartTimeOfSimulatedRaceTextBox.setEnabled(simulateWithStartTimeNowCheckBox.getValue());
offsetToStartTimeOfSimulatedRaceTextBox.setFocus(simulateWithStartTimeNowCheckBox.getValue());
}
});
final FlowPanel simulateAsLiveRacePanel = new FlowPanel();
simulateAsLiveRacePanel.add(simulateWithStartTimeNowCheckBox);
final Label offsetToStartLabel = new Label(stringMessages.simulateWithOffset());
final HorizontalPanel simulateWithOffsetPanel = new HorizontalPanel();
simulateWithOffsetPanel.add(offsetToStartLabel);
simulateWithOffsetPanel.add(offsetToStartTimeOfSimulatedRaceTextBox);
add(simulateAsLiveRacePanel);
add(simulateWithOffsetPanel);
}
public boolean isSimulate() {
return simulateWithStartTimeNowCheckBox.getValue();
}
public String getOffsetToStartTimeInMinutes() {
return offsetToStartTimeOfSimulatedRaceTextBox.getValue();
}
}
java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/adminconsole/TracTracEventManagementPanel.java
+4 -32
View File
@@ -23,7 +23,6 @@ import com.google.gwt.user.client.ui.CaptionPanel;
import com.google.gwt.user.client.ui.CheckBox;
import com.google.gwt.user.client.ui.FlexTable;
import com.google.gwt.user.client.ui.FlexTable.FlexCellFormatter;
import com.google.gwt.user.client.ui.FlowPanel;
import com.google.gwt.user.client.ui.Grid;
import com.google.gwt.user.client.ui.HTMLTable.ColumnFormatter;
import com.google.gwt.user.client.ui.HorizontalPanel;
@@ -85,7 +84,6 @@ public class TracTracEventManagementPanel extends AbstractEventManagementPanel {
private LabeledAbstractFilterablePanel<TracTracRaceRecordDTO> racesFilterablePanel;
private FlushableCellTable<TracTracRaceRecordDTO> racesTable;
private static final String ZERO_AS_STRING = "0";
public TracTracEventManagementPanel(final SailingServiceAsync sailingService, ErrorReporter errorReporter,
RegattaRefresher regattaRefresher, StringMessages stringMessages) {
@@ -275,32 +273,7 @@ public class TracTracEventManagementPanel extends AbstractEventManagementPanel {
correctWindCheckBox.setWordWrap(false);
correctWindCheckBox.setValue(Boolean.TRUE);
final TextBox offsetToStartTimeOfSimulatedRaceTextBox = new TextBox();
offsetToStartTimeOfSimulatedRaceTextBox.setWidth("40px");
offsetToStartTimeOfSimulatedRaceTextBox.setEnabled(false);
offsetToStartTimeOfSimulatedRaceTextBox.setValue(ZERO_AS_STRING);
final CheckBox simulateWithStartTimeNowCheckBox = new CheckBox(stringMessages.simulateAsLiveRace());
simulateWithStartTimeNowCheckBox.ensureDebugId("SimulateWithStartTimeNowCheckBox");
simulateWithStartTimeNowCheckBox.setWordWrap(false);
simulateWithStartTimeNowCheckBox.setValue(Boolean.FALSE);
simulateWithStartTimeNowCheckBox.addClickHandler(new ClickHandler() {
@Override
public void onClick(ClickEvent event) {
offsetToStartTimeOfSimulatedRaceTextBox.setEnabled(simulateWithStartTimeNowCheckBox.getValue());
offsetToStartTimeOfSimulatedRaceTextBox.setFocus(simulateWithStartTimeNowCheckBox.getValue());
}
});
final FlowPanel simulateAsLiveRacePanel = new FlowPanel();
simulateAsLiveRacePanel.add(simulateWithStartTimeNowCheckBox);
final Label offsetToStartLabel = new Label(stringMessages.simulateWithOffset());
final HorizontalPanel simulateWithOffsetPanel = new HorizontalPanel();
simulateWithOffsetPanel.add(offsetToStartLabel);
simulateWithOffsetPanel.add(offsetToStartTimeOfSimulatedRaceTextBox);
final SimulationPanel simulationPanel = new SimulationPanel(stringMessages);
final CheckBox ignoreTracTracMarkPassingsCheckbox = new CheckBox(stringMessages.useInternalAlgorithm());
ignoreTracTracMarkPassingsCheckbox.setWordWrap(false);
ignoreTracTracMarkPassingsCheckbox.setValue(Boolean.FALSE);
@@ -308,8 +281,7 @@ public class TracTracEventManagementPanel extends AbstractEventManagementPanel {
layoutTable.setWidget(++row, 0, trackSettingsLabel);
layoutTable.setWidget(row, 1, trackWindCheckBox);
layoutTable.setWidget(++row, 1, correctWindCheckBox);
layoutTable.setWidget(++row, 1, simulateAsLiveRacePanel);
layoutTable.setWidget(++row, 1, simulateWithOffsetPanel);
layoutTable.setWidget(++row, 1, simulationPanel);
layoutTable.setWidget(++row, 1, ignoreTracTracMarkPassingsCheckbox);
// Filter
@@ -413,8 +385,8 @@ public class TracTracEventManagementPanel extends AbstractEventManagementPanel {
@Override
public void onClick(ClickEvent event) {
Duration offsetToStartTimeOfSimulatedRace = null;
if(simulateWithStartTimeNowCheckBox.getValue().booleanValue()) {
offsetToStartTimeOfSimulatedRace = getMillisecondsDurationFromMinutesAsString(offsetToStartTimeOfSimulatedRaceTextBox.getValue());
if (simulationPanel.isSimulate()) {
offsetToStartTimeOfSimulatedRace = getMillisecondsDurationFromMinutesAsString(simulationPanel.getOffsetToStartTimeInMinutes());
}
trackSelectedRaces(trackWindCheckBox.getValue(), correctWindCheckBox.getValue(),
offsetToStartTimeOfSimulatedRace, ignoreTracTracMarkPassingsCheckbox.getValue());
java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/server/SailingServiceImpl.java
+1 -1
View File
@@ -2173,7 +2173,7 @@ public class SailingServiceImpl extends ProxiedRemoteServiceServlet implements S
} else {
actualTimePoint = timePoint;
}
RaceDefinition race = trackedRace.getRace();
final RaceDefinition race = trackedRace.getRace();
int rank = 1;
final List<Competitor> competitorsFromBestToWorst = trackedRace.getCompetitorsFromBestToWorst(actualTimePoint);
for (Competitor competitor : competitorsFromBestToWorst) {
java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/server/SimulatorServiceImpl.java
+3 -11
View File
@@ -10,10 +10,7 @@ import java.util.Map.Entry;
import java.util.NavigableMap;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ScheduledExecutorService;
import java.util.logging.Logger;
import com.google.gwt.user.server.rpc.RemoteServiceServlet;
@@ -86,7 +83,7 @@ import com.sap.sse.common.Duration;
import com.sap.sse.common.TimePoint;
import com.sap.sse.common.Util;
import com.sap.sse.common.impl.MillisecondsTimePoint;
import com.sap.sse.util.impl.ThreadFactoryWithPriority;
import com.sap.sse.util.ThreadPoolUtil;
public class SimulatorServiceImpl extends RemoteServiceServlet implements SimulatorService {
@@ -107,12 +104,7 @@ public class SimulatorServiceImpl extends RemoteServiceServlet implements Simula
private SpeedWithBearing averageWind = null;
public SimulatorServiceImpl() {
final int THREAD_POOL_SIZE = Math.max(Runtime.getRuntime().availableProcessors()/2, 3);
Executor simulatorExecutor = new ThreadPoolExecutor(/* corePoolSize */THREAD_POOL_SIZE,
/* maximumPoolSize */THREAD_POOL_SIZE,
/* keepAliveTime */60, TimeUnit.SECONDS,
/* workQueue */new LinkedBlockingQueue<Runnable>(),
new ThreadFactoryWithPriority(Thread.NORM_PRIORITY, /* daemon */ true));
final ScheduledExecutorService simulatorExecutor = ThreadPoolUtil.INSTANCE.getDefaultForegroundTaskThreadPoolExecutor();
// TODO: initialize smart-future-cache for simulation-results and add to simulation-service
simulationService = SimulationServiceFactory.INSTANCE.getService(simulatorExecutor, null);
}
java/com.sap.sailing.server.trackfiles/src/com/sap/sailing/server/trackfiles/impl/TrackFileExporterImpl.java
+2 -3
View File
@@ -6,7 +6,6 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -18,7 +17,7 @@ import com.sap.sailing.domain.common.trackfiles.TrackFilesFormat;
import com.sap.sailing.domain.trackimport.FormatNotSupportedException;
import com.sap.sailing.domain.tracking.TrackedRace;
import com.sap.sailing.server.trackfiles.TrackFileExporter;
import com.sap.sse.util.impl.ThreadFactoryWithPriority;
import com.sap.sse.util.ThreadPoolUtil;
/**
* Export data to well-known formats such as GPX, KML etc. Internally, the RouteConverter Library is used.
@@ -56,7 +55,7 @@ public class TrackFileExporterImpl implements TrackFileExporter {
};
List<WriteRaceDataCallable> callables = new ArrayList<>();
ExecutorService executor = Executors.newCachedThreadPool(new ThreadFactoryWithPriority(Thread.NORM_PRIORITY, /* daemon */ true));
ExecutorService executor = ThreadPoolUtil.INSTANCE.getDefaultForegroundTaskThreadPoolExecutor();
List<String> errors = new ArrayList<>();
for (TrackedRace race : races) {
for (TrackFilesDataSource d : data) {
java/com.sap.sailing.server/src/com/sap/sailing/server/impl/RacingEventServiceImpl.java
+1 -2
View File
@@ -28,7 +28,6 @@ import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -605,7 +604,7 @@ public class RacingEventServiceImpl implements RacingEventService, ClearStateTes
operationExecutionListeners = new ConcurrentHashMap<>();
courseListeners = new ConcurrentHashMap<>();
persistentRegattasForRaceIDs = new ConcurrentHashMap<>();
Executor simulatorExecutor = ThreadPoolUtil.INSTANCE.getDefaultBackgroundTaskThreadPoolExecutor();
final ScheduledExecutorService simulatorExecutor = ThreadPoolUtil.INSTANCE.getDefaultBackgroundTaskThreadPoolExecutor();
// TODO: initialize smart-future-cache for simulation-results and add to simulation-service
simulationService = SimulationServiceFactory.INSTANCE.getService(simulatorExecutor, this);
this.raceLogReplicator = new RaceLogReplicatorAndNotifier(this);
java/com.sap.sailing.server/src/com/sap/sailing/server/simulation/SimulationServiceFactory.java
+2 -2
View File
@@ -1,6 +1,6 @@
package com.sap.sailing.server.simulation;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import com.sap.sailing.server.RacingEventService;
@@ -8,6 +8,6 @@ public interface SimulationServiceFactory {
static SimulationServiceFactory INSTANCE = new SimulationServiceFactoryImpl();
SimulationService getService(Executor executor, RacingEventService racingEventService);
SimulationService getService(ScheduledExecutorService executor, RacingEventService racingEventService);
}
java/com.sap.sailing.server/src/com/sap/sailing/server/simulation/SimulationServiceFactoryImpl.java
+2 -2
View File
@@ -1,12 +1,12 @@
package com.sap.sailing.server.simulation;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import com.sap.sailing.server.RacingEventService;
public class SimulationServiceFactoryImpl implements SimulationServiceFactory {
public SimulationService getService(Executor executor, RacingEventService racingEventService) {
public SimulationService getService(ScheduledExecutorService executor, RacingEventService racingEventService) {
return new SimulationServiceImpl(executor, racingEventService);
}
java/com.sap.sailing.server/src/com/sap/sailing/server/simulation/SimulationServiceImpl.java
+14 -22
View File
@@ -8,9 +8,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.TreeMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
@@ -59,13 +57,12 @@ import com.sap.sse.common.TimePoint;
import com.sap.sse.common.impl.MillisecondsDurationImpl;
import com.sap.sse.common.impl.MillisecondsTimePoint;
import com.sap.sse.util.SmartFutureCache;
import com.sap.sse.util.impl.ThreadFactoryWithPriority;
public class SimulationServiceImpl implements SimulationService {
private static final Logger logger = Logger.getLogger(SimulationService.class.getName());
final private Executor executor;
final private ScheduledExecutorService executor;
final private SmartFutureCache<LegIdentifier, SimulationResults, SmartFutureCache.EmptyUpdateInterval> cache;
final private RacingEventService racingEventService;
final private ScheduledExecutorService scheduler;
@@ -73,9 +70,9 @@ public class SimulationServiceImpl implements SimulationService {
final private HashMap<RaceIdentifier, LegChangeListener> legListeners;
final private long WAIT_MILLIS = 20000; // milliseconds to wait until earliest cache-update for simulation
public SimulationServiceImpl(Executor executor, RacingEventService racingEventService) {
public SimulationServiceImpl(ScheduledExecutorService executor, RacingEventService racingEventService) {
this.executor = executor;
this.scheduler = Executors.newScheduledThreadPool(1, new ThreadFactoryWithPriority(Thread.NORM_PRIORITY, /* daemon */ true));
this.scheduler = executor;
this.racingEventService = racingEventService;
if (racingEventService != null) {
this.raceListeners = new HashMap<String, SimulationRaceListener>();
@@ -437,31 +434,26 @@ public class SimulationServiceImpl implements SimulationService {
Simulator simulator = new SimulatorImpl(simulationParameters);
Map<PathType, Path> result = new HashMap<PathType, Path>();
FutureTask<Path> taskOmniscient = null;
Future<Path> taskOmniscient = null;
if (simulationParameters.showOmniscient()) {
// schedule omniscient task
taskOmniscient = new FutureTask<Path>(() -> simulator.getPath(PathType.OMNISCIENT));
executor.execute(taskOmniscient);
taskOmniscient = executor.submit(() -> simulator.getPath(PathType.OMNISCIENT));
}
FutureTask<Path> task1TurnerLeft = null;
FutureTask<Path> task1TurnerRight = null;
Future<Path> task1TurnerLeft = null;
Future<Path> task1TurnerRight = null;
if (simulationParameters.getLegType() != LegType.REACHING) {
// schedule 1-turner tasks
task1TurnerLeft = new FutureTask<Path>(() -> simulator.getPath(PathType.ONE_TURNER_LEFT));
task1TurnerRight = new FutureTask<Path>(() -> simulator.getPath(PathType.ONE_TURNER_RIGHT));
executor.execute(task1TurnerLeft);
executor.execute(task1TurnerRight);
task1TurnerLeft = executor.submit(() -> simulator.getPath(PathType.ONE_TURNER_LEFT));
task1TurnerRight = executor.submit(() -> simulator.getPath(PathType.ONE_TURNER_RIGHT));
}
FutureTask<Path> taskOpportunistLeft = null;
FutureTask<Path> taskOpportunistRight = null;
Future<Path> taskOpportunistLeft = null;
Future<Path> taskOpportunistRight = null;
if (simulationParameters.showOpportunist()) {
// schedule opportunist tasks (which depend on 1-turner results)
taskOpportunistLeft = new FutureTask<Path>(() -> simulator.getPath(PathType.OPPORTUNIST_LEFT));
taskOpportunistRight = new FutureTask<Path>(() -> simulator.getPath(PathType.OPPORTUNIST_RIGHT));
executor.execute(taskOpportunistLeft);
executor.execute(taskOpportunistRight);
taskOpportunistLeft = executor.submit(() -> simulator.getPath(PathType.OPPORTUNIST_LEFT));
taskOpportunistRight = executor.submit(() -> simulator.getPath(PathType.OPPORTUNIST_RIGHT));
}
Path path1TurnerLeft = null;
java/com.sap.sse/src/com/sap/sse/util/SmartFutureCache.java
+45 -3
View File
@@ -13,6 +13,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -23,6 +24,11 @@ import com.sap.sse.common.Util.Pair;
import com.sap.sse.concurrent.LockUtil;
import com.sap.sse.concurrent.NamedReentrantReadWriteLock;
import com.sap.sse.util.SmartFutureCache.UpdateInterval;
import com.sap.sse.util.impl.HasTracingGet;
import com.sap.sse.util.impl.HasTracingGetImpl;
import com.sap.sse.util.impl.KnowsExecutor;
import com.sap.sse.util.impl.KnowsExecutorAndTracingGet;
import com.sap.sse.util.impl.KnowsExecutorAndTracingGetImpl;
/**
* A cache for which a background update can be triggered. Readers can decide whether they want to wait for any ongoing
@@ -188,6 +194,10 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
return callable.call();
}
@Override
public String toString() {
return callable==null?"null":callable.toString();
}
}
/**
@@ -201,7 +211,9 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
* @author Axel Uhl (D043530)
*
*/
private class FutureTaskWithCancelBlocking extends FutureTask<V> implements Callable<V> {
private class FutureTaskWithCancelBlocking extends FutureTask<V> implements KnowsExecutor, Callable<V> {
private final KnowsExecutorAndTracingGet<V> tracingGetHelper;
private boolean runningAndReadUpdateInterval;
private final K key;
@@ -228,6 +240,7 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
public FutureTaskWithCancelBlocking(SettableCallable<V> callable, final K key, final U updateInterval,
final boolean callerWaitsSynchronouslyForResult, final Thread callerThread) {
super(callable);
tracingGetHelper = new KnowsExecutorAndTracingGetImpl<>();
callable.setCallable(this);
this.gettingThreads = new HashSet<>();
this.key = key;
@@ -274,7 +287,7 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
}
}
}
return super.get();
return tracingGetHelper.callGetAndTraceAfterEachTimeout(this);
}
@Override
@@ -363,6 +376,16 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
public U getUpdateInterval() {
return updateInterval;
}
@Override
public void setExecutorThisTaskIsScheduledFor(ThreadPoolExecutor executorThisTaskIsScheduledFor) {
tracingGetHelper.setExecutorThisTaskIsScheduledFor(executorThisTaskIsScheduledFor);
}
@Override
public String toString() {
return getClass().getName()+" [key="+key+", updateInterval="+updateInterval+", cacheUpdateComputer="+cacheUpdateComputer+"]";
}
}
public SmartFutureCache(CacheUpdater<K, V, U> cacheUpdateComputer, String nameForLocks) {
@@ -507,14 +530,33 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
notifyAll();
}
}
public String toString() {
final String result;
synchronized (this) {
if (isSet) {
result = value==null?"null":value.toString();
} else {
result = "unset";
}
}
return result;
}
}
private static class TransitiveFuture<T> implements Future<T> {
private final SettableFuture<Future<T>> future;
private final HasTracingGet<T> tracingGetHelper;
protected TransitiveFuture(SettableFuture<Future<T>> future) {
super();
this.future = future;
this.tracingGetHelper = new HasTracingGetImpl<T>() {
@Override
protected String getAdditionalTraceInfo() {
return "transitive future "+TransitiveFuture.this.future.toString();
}
};
}
@Override
@@ -546,7 +588,7 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
@Override
public T get() throws InterruptedException, ExecutionException {
return future.get().get();
return tracingGetHelper.callGetAndTraceAfterEachTimeout(this);
}
@Override
java/com.sap.sse/src/com/sap/sse/util/ThreadPoolUtil.java
+4 -2
View File
@@ -52,7 +52,7 @@ public interface ThreadPoolUtil {
*
* The same configuration as for {@link #getDefaultBackgroundTaskThreadPoolExecutor()} is used.
*/
ScheduledExecutorService createBackgroundTaskThreadPoolExecutor();
ScheduledExecutorService createBackgroundTaskThreadPoolExecutor(String name);
/**
* In case an application module really requires its own thread pool instead of
@@ -74,5 +74,7 @@ public interface ThreadPoolUtil {
*
* The same configuration as for {@link #getDefaultForegroundTaskThreadPoolExecutor()} is used.
*/
ScheduledExecutorService createForegroundTaskThreadPoolExecutor();
ScheduledExecutorService createForegroundTaskThreadPoolExecutor(String name);
ScheduledExecutorService createForegroundTaskThreadPoolExecutor(int size, String name);
}
java/com.sap.sse/src/com/sap/sse/util/impl/FutureTaskWithTracingGet.java
+37
View File
@@ -0,0 +1,37 @@
package com.sap.sse.util.impl;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
public class FutureTaskWithTracingGet<V> extends FutureTask<V> {
private final HasTracingGet<V> hasTracingGet;
private final String traceInfo;
public FutureTaskWithTracingGet(String traceInfo, Callable<V> callable) {
super(callable);
this.traceInfo = traceInfo;
hasTracingGet = createHasTracingGet();
}
public FutureTaskWithTracingGet(String traceInfo, Runnable runnable, V result) {
super(runnable, result);
this.traceInfo = traceInfo;
hasTracingGet = createHasTracingGet();
}
@Override
public V get() throws InterruptedException, ExecutionException {
return hasTracingGet.callGetAndTraceAfterEachTimeout(this);
}
private HasTracingGet<V> createHasTracingGet() {
return new HasTracingGetImpl<V>() {
@Override
protected String getAdditionalTraceInfo() {
return traceInfo;
}
};
}
}
java/com.sap.sse/src/com/sap/sse/util/impl/HasTracingGet.java
+16
View File
@@ -0,0 +1,16 @@
package com.sap.sse.util.impl;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
public interface HasTracingGet<V> {
/**
* Keeps calling {@link Future#get(long, java.util.concurrent.TimeUnit)} with a timeout until a result is available
* or an exception other than a {@link TimeoutException} is thrown. When a timeout occurs, some trace output is
* written to the log. The method is guaranteed not to call {@link Future#get()} and therefore can be used in an
* implementation of a {@link Future#get()} method without causing an endless recursion as long as
* {@link Future#get(long, java.util.concurrent.TimeUnit)} is not calling this method.
*/
V callGetAndTraceAfterEachTimeout(Future<V> future) throws InterruptedException, ExecutionException;
}
java/com.sap.sse/src/com/sap/sse/util/impl/HasTracingGetImpl.java
+37
View File
@@ -0,0 +1,37 @@
package com.sap.sse.util.impl;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.logging.Level;
import java.util.logging.Logger;
import com.sap.sse.common.Duration;
/**
* Can be used with a {@link Future} task and outputs trace messages to the log after {@link #MILLIS_AFTER_WHICH_TO_TRACE_NON_RETURNING_GET} milliseconds
* of a {@link #get()} call not returning. The {@link #getAdditionalTraceInfo()} method must be implemented by subclasses to provide some meaningful
* hint as to which object is currently timing out.
*
* @author Axel Uhl (d043530)
*
* @param <V>
*/
public abstract class HasTracingGetImpl<V> implements HasTracingGet<V> {
private static final Logger logger = Logger.getLogger(HasTracingGetImpl.class.getName());
private final long MILLIS_AFTER_WHICH_TO_TRACE_NON_RETURNING_GET = Duration.ONE_SECOND.times(5).asMillis();
public V callGetAndTraceAfterEachTimeout(Future<V> future) throws InterruptedException, ExecutionException {
while (true) {
try {
return future.get(MILLIS_AFTER_WHICH_TO_TRACE_NON_RETURNING_GET, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
logger.log(Level.INFO, "Timeout waiting for future task "+future+
" (retrying); "+ getAdditionalTraceInfo(), e);
}
}
}
protected abstract String getAdditionalTraceInfo();
}
java/com.sap.sse/src/com/sap/sse/util/impl/KnowsExecutor.java
+14
View File
@@ -0,0 +1,14 @@
package com.sap.sse.util.impl;
import java.util.concurrent.ThreadPoolExecutor;
/**
* A {@link ThreadPoolExecutor} can be set on this object. This is used, in particular, by the
* {@link NamedTracingScheduledThreadPoolExecutor} whenever a task of this type is enqueued.
*
* @author Axel Uhl (d043530)
*
*/
public interface KnowsExecutor {
void setExecutorThisTaskIsScheduledFor(ThreadPoolExecutor executorThisTaskIsScheduledFor);
}
java/com.sap.sse/src/com/sap/sse/util/impl/KnowsExecutorAndTracingGet.java
+17
View File
@@ -0,0 +1,17 @@
package com.sap.sse.util.impl;
import java.util.concurrent.Future;
import com.sap.sse.util.ThreadPoolUtil;
/**
* Can be used with a {@link Future} task and outputs trace messages to the log after {@link #MILLIS_AFTER_WHICH_TO_TRACE_NON_RETURNING_GET} milliseconds
* of a {@link #get()} call not returning. When used in conjunction with a {@link NamedTracingScheduledThreadPoolExecutor} such as
* the ones returned by {@link ThreadPoolUtil}, the task will be
*
* @author Axel Uhl (d043530)
*
* @param <V>
*/
public interface KnowsExecutorAndTracingGet<V> extends KnowsExecutor, HasTracingGet<V> {
}
java/com.sap.sse/src/com/sap/sse/util/impl/KnowsExecutorAndTracingGetImpl.java
+31
View File
@@ -0,0 +1,31 @@
package com.sap.sse.util.impl;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import com.sap.sse.util.ThreadPoolUtil;
/**
* Can be used with a {@link Future} task and outputs trace messages to the log after
* {@link #MILLIS_AFTER_WHICH_TO_TRACE_NON_RETURNING_GET} milliseconds of a {@link #get()} call not returning. When used
* in conjunction with a {@link NamedTracingScheduledThreadPoolExecutor} such as the ones returned by
* {@link ThreadPoolUtil}, the task will be
*
* @author Axel Uhl (d043530)
*
* @param <V>
*/
public class KnowsExecutorAndTracingGetImpl<V> extends HasTracingGetImpl<V> implements KnowsExecutorAndTracingGet<V> {
private ThreadPoolExecutor executorThisTaskIsScheduledFor;
@Override
public void setExecutorThisTaskIsScheduledFor(ThreadPoolExecutor executorThisTaskIsScheduledFor) {
this.executorThisTaskIsScheduledFor = executorThisTaskIsScheduledFor;
}
@Override
protected String getAdditionalTraceInfo() {
return executorThisTaskIsScheduledFor == null ? "not scheduled"
: ("scheduled with executor " + executorThisTaskIsScheduledFor);
}
}
java/com.sap.sse/src/com/sap/sse/util/impl/NamedTracingScheduledThreadPoolExecutor.java
+164
View File
@@ -0,0 +1,164 @@
package com.sap.sse.util.impl;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.RunnableScheduledFuture;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import com.sap.sse.common.Duration;
import com.sap.sse.common.TimePoint;
import com.sap.sse.common.impl.MillisecondsTimePoint;
import com.sap.sse.concurrent.ConcurrentWeakHashMap;
/**
* A scheduled thread pool executor with a name that produces {@link Future} tasks that write a trace message
* to the log after their {@link Future#get()} method hasn't produced a result for some time, then keep trying.
* The tasks are linked to this executor, also when a {@link KnowsExecutor} task is submitted to the {@link #execute(Runnable)}
* method, so that when they need to trace they
*
* @author Axel Uhl (d043530)
*
*/
public class NamedTracingScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {
private static final Logger logger = Logger.getLogger(NamedTracingScheduledThreadPoolExecutor.class.getName());
private final String name;
/**
* If logging level is at least {@link Level#FINE}, each runnable's start time point is recorded in the
* {@link #beforeExecute(Thread, Runnable)} method. If at this logging level, the
* {@link #afterExecute(Runnable, Throwable)} method calculates the runnable's duration, and if above the
* {@link #DURATION_LOGGING_THRESHOLD}, writes a log entry.
*/
private final ConcurrentHashMap<Runnable, TimePoint> startTimePoints = new ConcurrentHashMap<>();
/**
* When logging {@link Level#FINE}, the {@link #schedule(Runnable, long, TimeUnit)} variants produce wrappers
* around the runnables whose content is nulled when {@link #afterExecute(Runnable, Throwable)} is called. In order
* to be able to log long-running runnables correctly, this map stores the connection between the wrapping
* future and the runnable wrapped by it.
*/
private final ConcurrentWeakHashMap<ScheduledFuture<?>, Runnable> wrappedRunnables = new ConcurrentWeakHashMap<>();
private final Duration DURATION_LOGGING_THRESHOLD = Duration.ONE_SECOND.times(5);
public NamedTracingScheduledThreadPoolExecutor(String name, int corePoolSize, RejectedExecutionHandler handler) {
super(corePoolSize, handler);
this.name = name;
}
public NamedTracingScheduledThreadPoolExecutor(String name, int corePoolSize, ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, threadFactory, handler);
this.name = name;
}
public NamedTracingScheduledThreadPoolExecutor(String name, int corePoolSize, ThreadFactory threadFactory) {
super(corePoolSize, threadFactory);
this.name = name;
}
public NamedTracingScheduledThreadPoolExecutor(String name, int corePoolSize) {
super(corePoolSize);
this.name = name;
}
@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
return new ThreadPoolAwareRunnableScheduledFutureDelegate<>(this, task);
}
@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
return new ThreadPoolAwareRunnableScheduledFutureDelegate<>(this, task);
}
@Override
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new ThreadPoolAwareFutureTask<T>(this, runnable, value);
}
@Override
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new ThreadPoolAwareFutureTask<T>(this, callable);
}
@Override
public void execute(Runnable command) {
if (command instanceof KnowsExecutor) {
((KnowsExecutor) command).setExecutorThisTaskIsScheduledFor(this);
}
super.execute(command);
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
super.beforeExecute(t, r);
if (logger.isLoggable(Level.FINE)) {
startTimePoints.put(r, MillisecondsTimePoint.now());
}
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
if (logger.isLoggable(Level.FINE)) {
final TimePoint end = MillisecondsTimePoint.now();
final TimePoint startTime = startTimePoints.remove(r);
Runnable theRunnableToLog = wrappedRunnables.get(r);
if (theRunnableToLog == null) {
theRunnableToLog = r;
}
if (startTime != null) {
final Duration duration = startTime.until(end);
if (duration.compareTo(DURATION_LOGGING_THRESHOLD) > 0) {
logger.fine("Runnable "+theRunnableToLog+" took "+duration+" to complete in executor "+this+(t==null?"":("It failed with exception "+t)));
}
} else {
logger.fine("Strange internal problem: had expected to find start time for "+r);
}
}
super.afterExecute(r, t);
}
@Override
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
final ScheduledFuture<?> result = super.schedule(command, delay, unit);
if (logger.isLoggable(Level.FINE)) {
// remember which runnable was wrapped by the future because it will have been nulled out when afterExecute is called
wrappedRunnables.put(result, command);
}
return result;
}
@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
final ScheduledFuture<?> result = super.scheduleAtFixedRate(command, initialDelay, period, unit);
if (logger.isLoggable(Level.FINE)) {
// remember which runnable was wrapped by the future because it will have been nulled out when afterExecute is called
wrappedRunnables.put(result, command);
}
return result;
}
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
final ScheduledFuture<?> result = super.scheduleWithFixedDelay(command, initialDelay, delay, unit);
if (logger.isLoggable(Level.FINE)) {
// remember which runnable was wrapped by the future because it will have been nulled out when afterExecute is called
wrappedRunnables.put(result, command);
}
return result;
}
@Override
public String toString() {
return super.toString() + "[name=" + name + "]";
}
}
java/com.sap.sse/src/com/sap/sse/util/impl/ThreadPoolAwareFutureTask.java
+62
View File
@@ -0,0 +1,62 @@
package com.sap.sse.util.impl;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.logging.Level;
import java.util.logging.Logger;
import com.sap.sse.util.ThreadPoolUtil;
/**
* A future task that outputs trace messages to the log after {@link #MILLIS_AFTER_WHICH_TO_TRACE_NON_RETURNING_GET} milliseconds
* of a {@link #get()} call not returning. When used in conjunction with a {@link NamedTracingScheduledThreadPoolExecutor} such as
* the ones returned by {@link ThreadPoolUtil}, the task will be updated by the executor when added to the queue. When the {@link #get()}
* method doesn't respond after the timeout period, the executor that scheduled this tasks, its name as well as its queue size
* are traced into the log.
*
* @author Axel Uhl (d043530)
*
* @param <V>
*/
public class ThreadPoolAwareFutureTask<V> extends FutureTask<V> implements KnowsExecutor {
private static final Logger logger = Logger.getLogger(ThreadPoolAwareFutureTask.class.getName());
private final KnowsExecutorAndTracingGet<V> getHelper = new KnowsExecutorAndTracingGetImpl<V>();
private final Object callableOrRunnableIfLoggingFine;
public ThreadPoolAwareFutureTask(ThreadPoolExecutor executor, Callable<V> callable) {
super(callable);
if (logger.isLoggable(Level.FINE)) {
callableOrRunnableIfLoggingFine = callable;
} else {
callableOrRunnableIfLoggingFine = null;
}
getHelper.setExecutorThisTaskIsScheduledFor(executor);
}
public ThreadPoolAwareFutureTask(ThreadPoolExecutor executor, Runnable runnable, V result) {
super(runnable, result);
if (logger.isLoggable(Level.FINE)) {
callableOrRunnableIfLoggingFine = runnable;
} else {
callableOrRunnableIfLoggingFine = null;
}
getHelper.setExecutorThisTaskIsScheduledFor(executor);
}
@Override
public void setExecutorThisTaskIsScheduledFor(ThreadPoolExecutor executorThisTaskIsScheduledFor) {
this.getHelper.setExecutorThisTaskIsScheduledFor(executorThisTaskIsScheduledFor);
}
@Override
public V get() throws InterruptedException, ExecutionException {
return getHelper.callGetAndTraceAfterEachTimeout(this);
}
@Override
public String toString() {
return super.toString()+(callableOrRunnableIfLoggingFine==null?"":(" "+callableOrRunnableIfLoggingFine.toString()));
}
}
java/com.sap.sse/src/com/sap/sse/util/impl/ThreadPoolAwareRunnableScheduledFutureDelegate.java
+53
View File
@@ -0,0 +1,53 @@
package com.sap.sse.util.impl;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.RunnableScheduledFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class ThreadPoolAwareRunnableScheduledFutureDelegate<V> extends KnowsExecutorAndTracingGetImpl<V> implements KnowsExecutorAndTracingGet<V>, RunnableScheduledFuture<V> {
private final RunnableScheduledFuture<V> future;
public ThreadPoolAwareRunnableScheduledFutureDelegate(ThreadPoolExecutor executor, RunnableScheduledFuture<V> future) {
this.future = future;
setExecutorThisTaskIsScheduledFor(executor);
}
public long getDelay(TimeUnit unit) {
return future.getDelay(unit);
}
public void run() {
future.run();
}
public boolean isPeriodic() {
return future.isPeriodic();
}
public boolean cancel(boolean mayInterruptIfRunning) {
return future.cancel(mayInterruptIfRunning);
}
public int compareTo(Delayed o) {
return future.compareTo(o);
}
public boolean isCancelled() {
return future.isCancelled();
}
public boolean isDone() {
return future.isDone();
}
public V get() throws InterruptedException, ExecutionException {
return callGetAndTraceAfterEachTimeout(future);
}
public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
return future.get(timeout, unit);
}
}
java/com.sap.sse/src/com/sap/sse/util/impl/ThreadPoolUtilImpl.java
+18 -13
View File
@@ -1,21 +1,18 @@
package com.sap.sse.util.impl;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import com.sap.sse.util.ThreadPoolUtil;
public class ThreadPoolUtilImpl implements ThreadPoolUtil {
private static final int REASONABLE_THREAD_POOL_SIZE = Math.max(Runtime.getRuntime().availableProcessors()/2, 3);
private static final int REASONABLE_THREAD_POOL_SIZE = Math.max(Runtime.getRuntime().availableProcessors()-1, 3);
private final ScheduledExecutorService defaultBackgroundTaskThreadPoolExecutor;
private final ScheduledExecutorService defaultForegroundTaskThreadPoolExecutor;
public ThreadPoolUtilImpl() {
defaultBackgroundTaskThreadPoolExecutor = Executors.newScheduledThreadPool(/* corePoolSize */ REASONABLE_THREAD_POOL_SIZE,
new ThreadFactoryWithPriority(Thread.NORM_PRIORITY-1, /* daemon */ true));
defaultForegroundTaskThreadPoolExecutor = Executors.newScheduledThreadPool(/* corePoolSize */ Math.max(Runtime.getRuntime().availableProcessors()-1, 6),
new ThreadFactoryWithPriority(Thread.NORM_PRIORITY, /* daemon */ true));
defaultBackgroundTaskThreadPoolExecutor = createBackgroundTaskThreadPoolExecutor("Default background executor");
defaultForegroundTaskThreadPoolExecutor = createForegroundTaskThreadPoolExecutor(2*REASONABLE_THREAD_POOL_SIZE, "Default foreground executor");
}
@Override
@@ -29,18 +26,26 @@ public class ThreadPoolUtilImpl implements ThreadPoolUtil {
}
@Override
public ScheduledExecutorService createBackgroundTaskThreadPoolExecutor() {
return createThreadPoolExecutor(Thread.NORM_PRIORITY-1);
public ScheduledExecutorService createBackgroundTaskThreadPoolExecutor(String name) {
return createThreadPoolExecutor(name, Thread.NORM_PRIORITY-1);
}
@Override
public ScheduledExecutorService createForegroundTaskThreadPoolExecutor() {
return createThreadPoolExecutor(Thread.NORM_PRIORITY);
public ScheduledExecutorService createForegroundTaskThreadPoolExecutor(String name) {
return createThreadPoolExecutor(name, Thread.NORM_PRIORITY);
}
private ScheduledExecutorService createThreadPoolExecutor(final int priority) {
return Executors.newScheduledThreadPool(/* corePoolSize */ REASONABLE_THREAD_POOL_SIZE,
new ThreadFactoryWithPriority(priority, /* daemon */ true));
@Override
public ScheduledExecutorService createForegroundTaskThreadPoolExecutor(int size, String name) {
return createThreadPoolExecutor(name, Thread.NORM_PRIORITY, size);
}
private ScheduledExecutorService createThreadPoolExecutor(String name, final int priority) {
return createThreadPoolExecutor(name, priority, /* corePoolSize */ REASONABLE_THREAD_POOL_SIZE);
}
private ScheduledExecutorService createThreadPoolExecutor(String name, final int priority, final int size) {
return new NamedTracingScheduledThreadPoolExecutor(name, /* corePoolSize */ size, new ThreadFactoryWithPriority(priority, /* daemon */ true));
}
@Override