tracing for Future.get

Change-Id: I5c0bf627c86eef180889fa769e7952de2a27afdd
This commit is contained in:
Axel Uhl committed 2016-09-27 17:17:33 +02:00
1 parent 69f46db026
commit ca14fe33f6
23 files changed
+407 -96

No files matched your search

@@ -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;
@@ -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) {
@@ -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);
@@ -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();
@@ -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;
@@ -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;
@@ -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);
}
@@ -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) {
@@ -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);
@@ -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);
}
@@ -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);
}
@@ -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;
@@ -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,9 @@ 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.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
@@ -201,7 +205,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 +234,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 +281,7 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
}
}
}
return super.get();
return tracingGetHelper.callGetAndTraceAfterEachTimeout(this);
}
@Override
@@ -363,6 +370,11 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
public U getUpdateInterval() {
return updateInterval;
}
@Override
public void setExecutorThisTaskIsScheduledFor(ThreadPoolExecutor executorThisTaskIsScheduledFor) {
tracingGetHelper.setExecutorThisTaskIsScheduledFor(executorThisTaskIsScheduledFor);
}
}
public SmartFutureCache(CacheUpdater<K, V, U> cacheUpdateComputer, String nameForLocks) {
@@ -51,8 +51,9 @@ public interface ThreadPoolUtil {
* <p>
*
* The same configuration as for {@link #getDefaultBackgroundTaskThreadPoolExecutor()} is used.
* @param name TODO
*/
ScheduledExecutorService createBackgroundTaskThreadPoolExecutor();
ScheduledExecutorService createBackgroundTaskThreadPoolExecutor(String name);
/**
* In case an application module really requires its own thread pool instead of
@@ -73,6 +74,7 @@ public interface ThreadPoolUtil {
* <p>
*
* The same configuration as for {@link #getDefaultForegroundTaskThreadPoolExecutor()} is used.
* @param name TODO
*/
ScheduledExecutorService createForegroundTaskThreadPoolExecutor();
ScheduledExecutorService createForegroundTaskThreadPoolExecutor(String name);
}
@@ -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;
}
};
}
}
@@ -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;
}
@@ -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 "+this+
"; "+ getAdditionalTraceInfo(), e);
}
}
}
protected abstract String getAdditionalTraceInfo();
}
@@ -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);
}
@@ -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> {
}
@@ -0,0 +1,32 @@
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 + " with queue size "
+ executorThisTaskIsScheduledFor.getQueue().size()));
}
}
@@ -0,0 +1,75 @@
package com.sap.sse.util.impl;
import java.util.concurrent.Callable;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.RunnableScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
/**
* 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 final String name;
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
public String toString() {
return "NamedTracingScheduledThreadPoolExecutor [name=" + name + "]";
}
}
@@ -0,0 +1,43 @@
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 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 final KnowsExecutorAndTracingGet<V> getHelper = new KnowsExecutorAndTracingGetImpl<V>();
public ThreadPoolAwareFutureTask(ThreadPoolExecutor executor, Callable<V> callable) {
super(callable);
getHelper.setExecutorThisTaskIsScheduledFor(executor);
}
public ThreadPoolAwareFutureTask(ThreadPoolExecutor executor, Runnable runnable, V result) {
super(runnable, result);
getHelper.setExecutorThisTaskIsScheduledFor(executor);
}
@Override
public void setExecutorThisTaskIsScheduledFor(ThreadPoolExecutor executorThisTaskIsScheduledFor) {
this.getHelper.setExecutorThisTaskIsScheduledFor(executorThisTaskIsScheduledFor);
}
@Override
public V get() throws InterruptedException, ExecutionException {
return getHelper.callGetAndTraceAfterEachTimeout(this);
}
}
@@ -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);
}
}
@@ -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("Default foreground executor");
}
@Override
@@ -29,18 +26,21 @@ 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));
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