mirror of
https://github.com/eclipse-sailing-analytics/sailing-analytics.git
synced 2026-11-14 22:43:57 +00:00
Merge branch 'master' into bug5217
This commit is contained in:
commit
09e8cd7c7a
25 files changed
+71
-236
No files matched your search
+1
-1
@@ -56,7 +56,7 @@ public class NOAAImporter extends DeclinationImporter {
|
||||
Date date = timePoint.asDate();
|
||||
Calendar calendar = new GregorianCalendar();
|
||||
calendar.setTime(date);
|
||||
URL url = new URL(QUERY_URL+"?lon1="+position.getLngDeg()+"&lat1="+position.getLatDeg()+"&startYear=" + calendar.get(Calendar.YEAR) + "&startMonth="
|
||||
URL url = new URL(QUERY_URL+"?key=zNEw7&lon1="+position.getLngDeg()+"&lat1="+position.getLatDeg()+"&startYear=" + calendar.get(Calendar.YEAR) + "&startMonth="
|
||||
+ (calendar.get(Calendar.MONTH) + 1) + "&startDay=" + calendar.get(Calendar.DAY_OF_MONTH)+"&resultFormat=xml");
|
||||
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
|
||||
conn.setRequestProperty("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8");
|
||||
|
||||
+1
-1
@@ -3073,7 +3073,7 @@ public class DomainObjectFactoryImpl implements DomainObjectFactory {
|
||||
});
|
||||
}, /* void result */ null);
|
||||
waiters.add(waiter);
|
||||
backgroundExecutor.execute(waiter);
|
||||
backgroundExecutor.execute(waiter); // OK to not associate with the current Subject here because we may not even have a session
|
||||
}
|
||||
logger.info("Done restoring races; restored " + i + " of " + count + " races");
|
||||
return new ConnectivityParametersLoadingResult() {
|
||||
|
||||
+1
-1
@@ -847,7 +847,7 @@ public class FixLoaderAndTracker implements TrackingDataLoader {
|
||||
loadingJobs.add(job);
|
||||
updateStatusAndProgress();
|
||||
}
|
||||
executor.execute(job);
|
||||
executor.execute(job); // nothing security-related happening here, and we may not have a Subject/session; no need to associate with a Subject
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
+2
-2
@@ -3135,7 +3135,7 @@ public class SailingServiceImpl extends ResultCachingProxiedRemoteServiceServlet
|
||||
}
|
||||
});
|
||||
resultFutures.put(competitorDTO, future);
|
||||
executor.execute(future);
|
||||
executor.execute(future); // security checks happen before; no need to associate future with Subject/session
|
||||
}
|
||||
for (Map.Entry<CompetitorDTO, FutureTask<CompetitorRaceDataDTO>> e : resultFutures.entrySet()) {
|
||||
CompetitorRaceDataDTO competitorData;
|
||||
@@ -3341,7 +3341,7 @@ public class SailingServiceImpl extends ResultCachingProxiedRemoteServiceServlet
|
||||
}
|
||||
return createManeuverDTOsForCompetitor(maneuvers, trackedRace, competitor);
|
||||
});
|
||||
executor.execute(future);
|
||||
executor.execute(future); // security checks happen before; no need to associate future with Subject
|
||||
futures.put(competitorDTO, future);
|
||||
}
|
||||
}
|
||||
|
||||
+4
-1
@@ -395,6 +395,7 @@ import com.sap.sse.shared.util.impl.UUIDHelper;
|
||||
import com.sap.sse.util.HttpUrlConnectionHelper;
|
||||
import com.sap.sse.util.ImageConverter;
|
||||
import com.sap.sse.util.ImageConverter.ImageWithMetadata;
|
||||
import com.sap.sse.util.ThreadPoolUtil;
|
||||
|
||||
public class SailingServiceWriteImpl extends SailingServiceImpl implements SailingServiceWrite {
|
||||
|
||||
@@ -1727,7 +1728,9 @@ public class SailingServiceWriteImpl extends SailingServiceImpl implements Saili
|
||||
exportTrackedRacesAndStartTracking, importOperationId);
|
||||
}
|
||||
};
|
||||
executor.execute(masterDataImportTask);
|
||||
// We need to convey the current user's credentials into the masterDataImportTask for default object ownership and permissions in case no
|
||||
// deviating username and password have been provided:
|
||||
executor.execute(ThreadPoolUtil.INSTANCE.associateWithSubjectIfAny(masterDataImportTask));
|
||||
return importOperationId;
|
||||
}
|
||||
|
||||
|
||||
+2
-2
@@ -881,14 +881,14 @@ public class LandscapeServiceImpl implements LandscapeService {
|
||||
if (replicaSet.getAutoScalingGroup() != null) {
|
||||
// run updating the auto-scaling group in the background; it's more time-critical now
|
||||
// to remove the old replicas from the target group
|
||||
ThreadPoolUtil.INSTANCE.getDefaultBackgroundTaskThreadPoolExecutor().execute(()->
|
||||
ThreadPoolUtil.INSTANCE.getDefaultBackgroundTaskThreadPoolExecutor().execute(ThreadPoolUtil.INSTANCE.associateWithSubjectIfAny(()->
|
||||
{
|
||||
try {
|
||||
getLandscape().updateAutoScalingGroupMinSize(replicaSet.getAutoScalingGroup(), oldAutoScalingGroupMinSize);
|
||||
} catch (InterruptedException | ExecutionException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
});
|
||||
}));
|
||||
} // else, the replica was started explicitly, without an auto-scaling group; in any case, all replicas still
|
||||
// on the old release will now be stopped:
|
||||
}
|
||||
|
||||
@@ -441,7 +441,7 @@ public class PolarDataMiner {
|
||||
}
|
||||
|
||||
public void raceFinishedTracking(final TrackedRace race) {
|
||||
processRacesThatFinishedLoadingExecutor.execute(()->{
|
||||
processRacesThatFinishedLoadingExecutor.execute(()->{ // no Subject association necessary here
|
||||
logger.info("All queued fixes for newly loaded race will process now. "
|
||||
+ (race.getRace() != null ? race.getRace().getName() : race.getRaceIdentifier().getRaceName()));
|
||||
for (final Competitor competitor : race.getRace().getCompetitors()) {
|
||||
|
||||
+5
-2
@@ -2305,6 +2305,7 @@ implements RacingEventService, ClearStateTestSupport, RegattaListener, Leaderboa
|
||||
final TimePoint now = MillisecondsTimePoint.now();
|
||||
if (notificationService != null && (lastNotificationForLeaderboard == null ||
|
||||
lastNotificationForLeaderboard.until(now).compareTo(HOW_LONG_BETWEEN_TWO_NOTIFICATIONS_FOR_SIMILAR_EVENT) >= 0)) {
|
||||
// no Subject association required for runnable here
|
||||
scheduler.execute(()->notificationService.notifyUserOnBoatClassWhenScoreCorrectionsAreAvailable(
|
||||
leaderboard.getBoatClass(), leaderboard));
|
||||
lastNotificationForLeaderboard = now;
|
||||
@@ -2320,13 +2321,13 @@ implements RacingEventService, ClearStateTestSupport, RegattaListener, Leaderboa
|
||||
final TimePoint now = MillisecondsTimePoint.now();
|
||||
if (notificationService != null && (!lastNotificationForCompetitor.containsKey(competitor) ||
|
||||
lastNotificationForCompetitor.get(competitor).until(now).compareTo(HOW_LONG_BETWEEN_TWO_NOTIFICATIONS_FOR_SIMILAR_EVENT) >= 0)) {
|
||||
// no Subject association required for runnable here
|
||||
scheduler.execute(()->notificationService.notifyUserOnCompetitorScoreCorrections(competitor, leaderboard));
|
||||
lastNotificationForCompetitor.put(competitor, now);
|
||||
}
|
||||
// a change to a single competitor also means a change to the leaderboard
|
||||
notifyForLeaderboardIfNotAlreadyNotifiedRecently();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private class PolarFixCacheUpdater extends AbstractRaceChangeListener {
|
||||
@@ -2396,8 +2397,8 @@ implements RacingEventService, ClearStateTestSupport, RegattaListener, Leaderboa
|
||||
public void startOfRaceChanged(TimePoint oldStartOfRace, TimePoint newStartOfRace) {
|
||||
// no replication action required; the update signaled by this call is implicit; for explicit updates
|
||||
// see raceTimesChanged(TimePoint, TimePoint, TimePoint).
|
||||
|
||||
if (newStartOfRace != null && newStartOfRace.after(MillisecondsTimePoint.now())) {
|
||||
// no Subject association required for runnable here
|
||||
scheduler.execute(()->
|
||||
// Notify interested users if the new start time is in the future
|
||||
notificationService.notifyUserOnBoatClassUpcomingRace(trackedRace.getRace().getBoatClass(),
|
||||
@@ -2415,6 +2416,7 @@ implements RacingEventService, ClearStateTestSupport, RegattaListener, Leaderboa
|
||||
// no action required; the update signaled by this call is implicit; the race log
|
||||
// updates that led to this change are replicated separately
|
||||
if (newFinishedTime != null && newFinishedTime.after(MillisecondsTimePoint.now().minus(Duration.ONE_HOUR))) {
|
||||
// no Subject association required for runnable here
|
||||
scheduler.execute(()->
|
||||
// Notify interested users:
|
||||
notificationService.notifyUserOnBoatClassRaceChangesStateToFinished(trackedRace.getRace().getBoatClass(), trackedRace,
|
||||
@@ -2485,6 +2487,7 @@ implements RacingEventService, ClearStateTestSupport, RegattaListener, Leaderboa
|
||||
if (last != null && last.getWaypoint() == trackedRace.getRace().getCourse().getLastWaypoint() &&
|
||||
trackedRace.getStatus().getStatus() != TrackedRaceStatusEnum.LOADING &&
|
||||
last.getTimePoint().after(MillisecondsTimePoint.now().minus(Duration.ONE_HOUR))) {
|
||||
// no Subject association required for runnable here
|
||||
scheduler.execute(() ->
|
||||
// Notify interested users:
|
||||
notificationService.notifyUserOnCompetitorPassesFinish(competitor, trackedRace,
|
||||
|
||||
-1
@@ -192,7 +192,6 @@ public class SailingNotificationServiceImpl implements SailingNotificationServic
|
||||
Leaderboard leaderboard, RaceColumn raceColumn, Fleet fleet) {
|
||||
doWithEvent(leaderboard, (event, leaderboardGroup) -> {
|
||||
mailQueue.addNotification(new NotificationSetNotification<BoatClass>(boatClass, boatClassResults) {
|
||||
|
||||
@Override
|
||||
protected NotificationMailTemplate getMailTemplate(BoatClass objectToNotifyAbout, Locale locale) {
|
||||
String raceDescription = calculateRaceDescription(locale, event, leaderboard, raceColumn, fleet);
|
||||
|
||||
+1
@@ -210,6 +210,7 @@ public class ManeuverAndWindImporter {
|
||||
i++;
|
||||
final int raceNumber = i;
|
||||
logger.info("Processing race nr. " + raceNumber + ": \"" + trackedRaceName + "\"");
|
||||
// no Subject association required for runnable here
|
||||
executorService.execute(() -> {
|
||||
try {
|
||||
importRace(regattaName, trackedRaceName, importStatistics, bearerToken);
|
||||
|
||||
-1
@@ -206,7 +206,6 @@ public abstract class AbstractGenericSerializableSettings extends AbstractSettin
|
||||
sb.append("; ");
|
||||
}
|
||||
sb.append(childSetting.getKey() + "=" + childSetting.getValue());
|
||||
|
||||
first = false;
|
||||
}
|
||||
sb.append("}");
|
||||
|
||||
+1
@@ -8,6 +8,7 @@ import java.util.concurrent.TimeUnit;
|
||||
import com.sap.sse.util.impl.ThreadFactoryWithPriority;
|
||||
|
||||
public class DataMiningExecutorService extends ThreadPoolExecutor {
|
||||
// FIXME bug4821: what about security session state with this executor? Where do we submit tasks to this executor, and what should their Shiro session state / Subject be?
|
||||
public DataMiningExecutorService(int corePoolSize) {
|
||||
super(corePoolSize, corePoolSize, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(corePoolSize),
|
||||
new ThreadFactoryWithPriority(Thread.NORM_PRIORITY, /* daemon */ true),
|
||||
|
||||
@@ -24,6 +24,7 @@ public class ExecutorMailQueue implements MailQueue {
|
||||
|
||||
@Override
|
||||
public void addNotification(MailNotification notification) {
|
||||
// no Subject association required for runnable here
|
||||
ThreadPoolUtil.INSTANCE.getDefaultBackgroundTaskThreadPoolExecutor()
|
||||
.execute(() -> notification.sendNotifications(mailServiceTracker.getService()));
|
||||
}
|
||||
|
||||
-55
@@ -1,55 +0,0 @@
|
||||
package com.sap.sse.security.test;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.Future;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import com.sap.sse.util.impl.NamedTracingScheduledThreadPoolExecutor;
|
||||
|
||||
public class InheritableThreadLocalAndThreadPoolTest {
|
||||
private static InheritableThreadLocal<Integer> defaultThreadLocal;
|
||||
private static InheritableThreadLocal<Integer> threadLocalWithChildValueInit;
|
||||
private NamedTracingScheduledThreadPoolExecutor executor;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
defaultThreadLocal = new InheritableThreadLocal<>();
|
||||
threadLocalWithChildValueInit = new InheritableThreadLocal<Integer>() {
|
||||
@Override
|
||||
protected Integer childValue(Integer parentValue) {
|
||||
return 42;
|
||||
}
|
||||
};
|
||||
executor = new NamedTracingScheduledThreadPoolExecutor("Test Executor", /* corePoolSize */ 1);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSimpleValue() throws InterruptedException, ExecutionException {
|
||||
defaultThreadLocal.set(14);
|
||||
final Callable<Integer> reader1 = ()->defaultThreadLocal.get();
|
||||
final Future<Integer> future1 = executor.submit(reader1);
|
||||
defaultThreadLocal.set(15);
|
||||
final Callable<Integer> reader2 = ()->defaultThreadLocal.get();
|
||||
final Future<Integer> future2 = executor.submit(reader2);
|
||||
assertEquals(14, future1.get().intValue());
|
||||
assertEquals(15, future2.get().intValue());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMissingInitialization() throws InterruptedException, ExecutionException {
|
||||
threadLocalWithChildValueInit.set(123);
|
||||
final Callable<Integer> reader1 = ()->threadLocalWithChildValueInit.get();
|
||||
final Future<Integer> future1 = executor.submit(reader1);
|
||||
final Runnable setter = ()->threadLocalWithChildValueInit.set(43);
|
||||
executor.execute(setter);
|
||||
final Callable<Integer> reader2 = ()->threadLocalWithChildValueInit.get();
|
||||
final Future<Integer> future2 = executor.submit(reader2);
|
||||
assertEquals(42, future1.get().intValue());
|
||||
assertEquals(42, future2.get().intValue());
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package com.sap.sse.test;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertSame;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.lang.ref.WeakReference;
|
||||
@@ -13,8 +14,6 @@ import java.util.logging.Logger;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import com.sap.sse.util.ThreadPoolUtil;
|
||||
|
||||
public class ThreadPoolWithThreadLocalsTest {
|
||||
|
||||
private static final Logger logger = Logger.getLogger(ThreadPoolWithThreadLocalsTest.class.getName());
|
||||
@@ -43,24 +42,11 @@ public class ThreadPoolWithThreadLocalsTest {
|
||||
testThreadReuseWithThreadLocal(() -> Executors.newFixedThreadPool(2));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testThreadReuseWithThreadLocalUsingNamedTracingScheduledThreadPoolExecutor() throws Exception {
|
||||
testThreadReuseWithThreadLocal(
|
||||
() -> ThreadPoolUtil.INSTANCE.createForegroundTaskThreadPoolExecutor("ThreadPoolWithThreadLocalsTest"));
|
||||
}
|
||||
|
||||
@Test(expected = ThreadLocalCollisionException.class)
|
||||
public void testThreadReuseWithInheritableThreadLocalUsingFixedThreadPool() throws Exception {
|
||||
testThreadReuseWithInheritableThreadLocal(() -> Executors.newFixedThreadPool(2));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testThreadReuseWithInheritableThreadLocalUsingNamedTracingScheduledThreadPoolExecutor()
|
||||
throws Exception {
|
||||
testThreadReuseWithInheritableThreadLocal(() -> ThreadPoolUtil.INSTANCE
|
||||
.createForegroundTaskThreadPoolExecutor("ThreadPoolWithInheritableThreadLocalsTest"));
|
||||
}
|
||||
|
||||
private void testThreadReuseWithThreadLocal(final Supplier<ExecutorService> executorFactory) throws Exception {
|
||||
final ExecutorService executor = executorFactory.get();
|
||||
final Thread thread1 = executor.submit(() -> {
|
||||
@@ -73,7 +59,7 @@ public class ThreadPoolWithThreadLocalsTest {
|
||||
while (thread1 != thread2) {
|
||||
thread2 = executor.submit(() -> Thread.currentThread()).get();
|
||||
}
|
||||
assertEquals(thread1, thread2);
|
||||
assertSame(thread1, thread2);
|
||||
if (containsThreadLocalWithValue(thread2, "top secret", threadLocalsField)) {
|
||||
throw new ThreadLocalCollisionException(
|
||||
"Reused Thread contains some old top secret value from ThreadLocal!");
|
||||
|
||||
@@ -27,7 +27,8 @@ Require-Bundle: com.sap.sse.common,
|
||||
org.json.simple;bundle-version="1.1.0",
|
||||
org.apache.httpcomponents.httpclient,
|
||||
org.apache.httpcomponents.httpcore;bundle-version="4.4.9",
|
||||
com.sun.jersey
|
||||
com.sun.jersey,
|
||||
org.apache.shiro.core
|
||||
Import-Package: javax.servlet;version="3.1.0",
|
||||
javax.servlet.http;version="3.1.0",
|
||||
javax.ws.rs.core;version="1.1.1",
|
||||
|
||||
@@ -20,6 +20,8 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.shiro.subject.Subject;
|
||||
|
||||
import com.sap.sse.common.Util.Pair;
|
||||
import com.sap.sse.concurrent.LockUtil;
|
||||
import com.sap.sse.concurrent.NamedReentrantReadWriteLock;
|
||||
@@ -206,7 +208,10 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
|
||||
* the Future hasn't been cancelled yet. To avoid that the Future is cancelled after the client has fetched it from
|
||||
* {@link SmartFutureCache#ongoingRecalculations}, the client can call {@link #dontCancel()} on this
|
||||
* future. After that, calls to {@link #cancel(boolean)} will return <code>false</code> immediately and the Future
|
||||
* will be executed as originally scheduled.
|
||||
* will be executed as originally scheduled.<p>
|
||||
*
|
||||
* A task of this sort will always {@link Subject#associateWith(Runnable) associate} any current {@link Subject}
|
||||
* to the task when it is executed.
|
||||
*
|
||||
* @author Axel Uhl (D043530)
|
||||
*
|
||||
@@ -239,7 +244,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);
|
||||
super(ThreadPoolUtil.INSTANCE.associateWithSubjectIfAny(callable));
|
||||
tracingGetHelper = new KnowsExecutorAndTracingGetImpl<>();
|
||||
callable.setCallable(this);
|
||||
this.gettingThreads = new HashSet<>();
|
||||
@@ -249,11 +254,6 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
|
||||
this.callerThread = callerThread;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<ThreadLocal<Object>, Object> getThreadLocalValuesToInherit() {
|
||||
return tracingGetHelper.getThreadLocalValuesToInherit();
|
||||
}
|
||||
|
||||
/**
|
||||
* Tries to update this task's update interval so as to re-used an already scheduled task for a given key.
|
||||
* The update can only be performed if the task hasn't read the update interval after it has been started.
|
||||
@@ -297,7 +297,6 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
|
||||
|
||||
@Override
|
||||
public V call() {
|
||||
tracingGetHelper.setInheritableThreadLocalValues();
|
||||
try {
|
||||
final U updateInterval;
|
||||
final Set<Thread> locksPropagatedFromGettingThreads;
|
||||
@@ -371,8 +370,6 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
|
||||
LockUtil.unpropagateLockSetFrom(callerThread);
|
||||
}
|
||||
gettingThreads.clear();
|
||||
tracingGetHelper.removeInheritableThreadLocalValues();
|
||||
tracingGetHelper.removeThreadLocalValues();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
// cache won't be updated
|
||||
@@ -663,6 +660,7 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
|
||||
// is too expensive here.
|
||||
// logger.finest("creating future task on cache "+nameForLocks+" for key "+key);
|
||||
ongoingRecalculations.put(key, future);
|
||||
// the FutureTaskWithCancelBlocking associates any current Subject with the task while it is executed
|
||||
recalculator.execute(future);
|
||||
return future;
|
||||
}
|
||||
|
||||
@@ -113,4 +113,8 @@ public interface ThreadPoolUtil {
|
||||
* by the exception's message.
|
||||
*/
|
||||
<T> List<Future<T>> invokeAllAndLogExceptions(ExecutorService executor, Level logLevel, String messageTemplate, Iterable<? extends Callable<T>> tasks);
|
||||
|
||||
Runnable associateWithSubjectIfAny(Runnable runnable);
|
||||
|
||||
<T> Callable<T> associateWithSubjectIfAny(Callable<T> callable);
|
||||
}
|
||||
@@ -1,6 +1,5 @@
|
||||
package com.sap.sse.util.impl;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
|
||||
/**
|
||||
@@ -12,5 +11,4 @@ import java.util.concurrent.ThreadPoolExecutor;
|
||||
*/
|
||||
public interface KnowsExecutor {
|
||||
void setExecutorThisTaskIsScheduledFor(ThreadPoolExecutor executorThisTaskIsScheduledFor);
|
||||
Map<ThreadLocal<Object>, Object> getThreadLocalValuesToInherit();
|
||||
}
|
||||
@@ -14,24 +14,4 @@ import com.sap.sse.util.ThreadPoolUtil;
|
||||
* @param <V>
|
||||
*/
|
||||
public interface KnowsExecutorAndTracingGet<V> extends KnowsExecutor, HasTracingGet<V> {
|
||||
/**
|
||||
* Assuming that the thread pool's thread has taken over and is about to run this task,
|
||||
* sets all inheritable thread local values for the executing thread. The values
|
||||
* are obtained through {@link #getThreadLocalValuesToInherit()}.
|
||||
*/
|
||||
void setInheritableThreadLocalValues();
|
||||
|
||||
/**
|
||||
* Removes / "unsets" the values for all thread locals whose values were memorized by
|
||||
* {@link #getThreadLocalValuesToInherit()}. Call this method at the end of the
|
||||
* call/run method of this task.
|
||||
*/
|
||||
void removeInheritableThreadLocalValues();
|
||||
|
||||
/**
|
||||
* Removes all values stored in {@link ThreadLocal}s of a thread. This should be called
|
||||
* after task execution to prevent illegal {@link ThreadLocal} value reuse when using
|
||||
* thread pools.
|
||||
*/
|
||||
public void removeThreadLocalValues();
|
||||
}
|
||||
@@ -1,17 +1,10 @@
|
||||
package com.sap.sse.util.impl;
|
||||
|
||||
import java.lang.ref.WeakReference;
|
||||
import java.lang.reflect.Field;
|
||||
import java.lang.reflect.Method;
|
||||
import java.util.Collections;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.WeakHashMap;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.shiro.SecurityUtils;
|
||||
import org.apache.shiro.subject.Subject;
|
||||
|
||||
import com.sap.sse.util.ThreadPoolUtil;
|
||||
|
||||
@@ -26,97 +19,13 @@ import com.sap.sse.util.ThreadPoolUtil;
|
||||
* @param <V>
|
||||
*/
|
||||
public class KnowsExecutorAndTracingGetImpl<V> extends HasTracingGetImpl<V> implements KnowsExecutorAndTracingGet<V> {
|
||||
private static final Logger logger = Logger.getLogger(KnowsExecutorAndTracingGetImpl.class.getName());
|
||||
private static Field inheritableThreadLocalsField;
|
||||
private static Method childValueMethod;
|
||||
private static Field threadLocalsField;
|
||||
private ThreadPoolExecutor executorThisTaskIsScheduledFor;
|
||||
private final WeakHashMap<ThreadLocal<Object>, Object> threadLocalValuesToInherit;
|
||||
|
||||
static {
|
||||
try {
|
||||
inheritableThreadLocalsField = Thread.class.getDeclaredField("inheritableThreadLocals");
|
||||
inheritableThreadLocalsField.setAccessible(true);
|
||||
childValueMethod = InheritableThreadLocal.class.getDeclaredMethod("childValue", Object.class);
|
||||
childValueMethod.setAccessible(true);
|
||||
threadLocalsField = Thread.class.getDeclaredField("threadLocals");
|
||||
threadLocalsField.setAccessible(true);
|
||||
} catch (NoSuchFieldException | SecurityException | NoSuchMethodException e) {
|
||||
logger.log(Level.SEVERE,
|
||||
"Problem finding field inheritableThreadLocals; will be unable to forward InheritableThreadLocal values to thread pool threads",
|
||||
e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Copies the values of all {@link InheritableThreadLocal} to the {@link #threadLocalValuesToInherit} map so
|
||||
* that they can be set just before this task is run.
|
||||
* Captures any Shiro {@link Subject} through the use of {@link SecurityUtils#getSubject()}. If an
|
||||
* {@link IllegalStateException} is raised, the {@link #subject} is set to {@code null}.
|
||||
*/
|
||||
public KnowsExecutorAndTracingGetImpl() {
|
||||
final WeakHashMap<ThreadLocal<Object>, Object> threadLocalValuesToInheritValue = new WeakHashMap<>();
|
||||
final Thread currentThread = Thread.currentThread();
|
||||
handleThreadLocals(currentThread, inheritableThreadLocalsField, (key, value) -> {
|
||||
threadLocalValuesToInheritValue.put(key, value);
|
||||
});
|
||||
threadLocalValuesToInherit = threadLocalValuesToInheritValue;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setInheritableThreadLocalValues() {
|
||||
for (Entry<ThreadLocal<Object>, Object> e : threadLocalValuesToInherit.entrySet()) {
|
||||
e.getKey().set(e.getValue());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void removeInheritableThreadLocalValues() {
|
||||
for (Entry<ThreadLocal<Object>, Object> e : threadLocalValuesToInherit.entrySet()) {
|
||||
e.getKey().remove();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void removeThreadLocalValues() {
|
||||
handleThreadLocals(Thread.currentThread(), threadLocalsField, (key, value) -> key.remove());
|
||||
}
|
||||
|
||||
private void handleThreadLocals(final Thread thread, final Field threadLocalSourceField,
|
||||
final BiConsumer<ThreadLocal<Object>, Object> valueHandler) {
|
||||
try {
|
||||
final Object threadLocals = threadLocalSourceField.get(thread);
|
||||
if (threadLocals != null) {
|
||||
final Method expungeStaleEntries = threadLocals.getClass().getDeclaredMethod("expungeStaleEntries");
|
||||
expungeStaleEntries.setAccessible(true);
|
||||
expungeStaleEntries.invoke(threadLocals);
|
||||
final Field tableField = threadLocals.getClass().getDeclaredField("table");
|
||||
tableField.setAccessible(true);
|
||||
final Object[] table = (Object[]) tableField.get(threadLocals);
|
||||
for (int j = 0; j < table.length; j++) {
|
||||
final WeakReference<?> entry = (WeakReference<?>) table[j];
|
||||
if (entry != null) {
|
||||
@SuppressWarnings("unchecked")
|
||||
final ThreadLocal<Object> key = (ThreadLocal<Object>) entry.get();
|
||||
if (key != null) {
|
||||
final Field valueField = entry.getClass().getDeclaredField("value");
|
||||
valueField.setAccessible(true);
|
||||
final Object value = key instanceof InheritableThreadLocal
|
||||
? childValueMethod.invoke(key, valueField.get(entry))
|
||||
: key.get();
|
||||
valueHandler.accept(key, value);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
logger.log(Level.SEVERE,
|
||||
"Problem accessing field inheritableThreadLocals; will be unable to forward InheritableThreadLocal values to thread pool threads",
|
||||
e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<ThreadLocal<Object>, Object> getThreadLocalValuesToInherit() {
|
||||
return Collections.unmodifiableMap(threadLocalValuesToInherit);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
package com.sap.sse.util.impl;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.FutureTask;
|
||||
@@ -28,7 +27,7 @@ public class ThreadPoolAwareFutureTask<V> extends FutureTask<V> implements Knows
|
||||
private final Object delegate; // either a Callable or a Runnable whose type may or may not conform to the KnowsExecutor interface
|
||||
|
||||
public ThreadPoolAwareFutureTask(ThreadPoolExecutor executor, Callable<V> callable) {
|
||||
super(callable);
|
||||
super(ThreadPoolUtil.INSTANCE.associateWithSubjectIfAny(callable));
|
||||
delegate = callable;
|
||||
if (logger.isLoggable(Level.FINE)) {
|
||||
callableOrRunnableIfLoggingFine = callable;
|
||||
@@ -39,7 +38,7 @@ public class ThreadPoolAwareFutureTask<V> extends FutureTask<V> implements Knows
|
||||
}
|
||||
|
||||
public ThreadPoolAwareFutureTask(ThreadPoolExecutor executor, Runnable runnable, V result) {
|
||||
super(runnable, result);
|
||||
super(ThreadPoolUtil.INSTANCE.associateWithSubjectIfAny(runnable), result);
|
||||
delegate = runnable;
|
||||
if (logger.isLoggable(Level.FINE)) {
|
||||
callableOrRunnableIfLoggingFine = runnable;
|
||||
@@ -51,13 +50,7 @@ public class ThreadPoolAwareFutureTask<V> extends FutureTask<V> implements Knows
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
getHelper.setInheritableThreadLocalValues();
|
||||
try {
|
||||
super.run();
|
||||
} finally {
|
||||
getHelper.removeInheritableThreadLocalValues();
|
||||
getHelper.removeThreadLocalValues();
|
||||
}
|
||||
super.run();
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -68,11 +61,6 @@ public class ThreadPoolAwareFutureTask<V> extends FutureTask<V> implements Knows
|
||||
this.getHelper.setExecutorThisTaskIsScheduledFor(executorThisTaskIsScheduledFor);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<ThreadLocal<Object>, Object> getThreadLocalValuesToInherit() {
|
||||
return getHelper.getThreadLocalValuesToInherit();
|
||||
}
|
||||
|
||||
@Override
|
||||
public V get() throws InterruptedException, ExecutionException {
|
||||
return getHelper.callGetAndTraceAfterEachTimeout(this);
|
||||
|
||||
+1
-7
@@ -20,13 +20,7 @@ public class ThreadPoolAwareRunnableScheduledFutureDelegate<V> extends KnowsExec
|
||||
}
|
||||
|
||||
public void run() {
|
||||
setInheritableThreadLocalValues();
|
||||
try {
|
||||
future.run();
|
||||
} finally {
|
||||
removeInheritableThreadLocalValues();
|
||||
removeThreadLocalValues();
|
||||
}
|
||||
future.run();
|
||||
}
|
||||
|
||||
public boolean isPeriodic() {
|
||||
|
||||
@@ -2,6 +2,7 @@ package com.sap.sse.util.impl;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
@@ -10,6 +11,10 @@ import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.shiro.SecurityUtils;
|
||||
import org.apache.shiro.UnavailableSecurityManagerException;
|
||||
import org.apache.shiro.subject.Subject;
|
||||
|
||||
import com.sap.sse.common.Util;
|
||||
import com.sap.sse.util.ThreadPoolUtil;
|
||||
|
||||
@@ -105,4 +110,24 @@ public class ThreadPoolUtilImpl implements ThreadPoolUtil {
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
private Optional<Subject> getSubjectOrNull() {
|
||||
Optional<Subject> mySubject;
|
||||
try {
|
||||
mySubject = Optional.of(SecurityUtils.getSubject());
|
||||
} catch (IllegalStateException | UnavailableSecurityManagerException e) {
|
||||
mySubject = Optional.empty();
|
||||
}
|
||||
return mySubject;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Runnable associateWithSubjectIfAny(Runnable runnable) {
|
||||
return getSubjectOrNull().map(subject->subject.associateWith(runnable)).orElse(runnable);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T> Callable<T> associateWithSubjectIfAny(Callable<T> callable) {
|
||||
return getSubjectOrNull().map(subject->subject.associateWith(callable)).orElse(callable);
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -78,7 +78,7 @@ JAVA_BINARY="$JAVA_HOME/bin/java"
|
||||
JAVA_VERSION_OUTPUT=$("$JAVA_BINARY" -version 2>&1)
|
||||
JAVA_VERSION=$(echo "$JAVA_VERSION_OUTPUT" | sed 's/^.* version "\(.*\)\.\(.*\)\..*".*$/\1.\2/; 1q')
|
||||
export JAVA_11_LOGGING_ARGS="-Xlog:gc+ergo*=trace:file=logs/gc_ergo.log:time:filecount=10,filesize=10000000 -Xlog:gc*:file=logs/gc.log:time:filecount=10,filesize=100000"
|
||||
export JAVA_11_ARGS="-Dosgi.java.profile=file://`pwd`/JavaSE-11.profile --add-modules=ALL-SYSTEM -Djavax.xml.bind.JAXBContextFactory=com.sun.xml.bind.v2.ContextFactory -XX:ThreadPriorityPolicy=1 -XX:+UnlockExperimentalVMOptions -XX:+UseZGC ${JAVA_11_LOGGING_ARGS} --illegal-access=permit"
|
||||
export JAVA_11_ARGS="-Dosgi.java.profile=file://`pwd`/JavaSE-11.profile --add-modules=ALL-SYSTEM -Djavax.xml.bind.JAXBContextFactory=com.sun.xml.bind.v2.ContextFactory -XX:ThreadPriorityPolicy=1 -XX:+UnlockExperimentalVMOptions -XX:+UseZGC ${JAVA_11_LOGGING_ARGS}"
|
||||
export JAVA_8_LOGGING_ARGS="-XX:+PrintAdaptiveSizePolicy -XX:+PrintGCTimeStamps -XX:+PrintGCDetails -Xloggc:logs/gc.log -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=100M"
|
||||
echo JAVA_VERSION detected: $JAVA_VERSION >&2
|
||||
JAVA_MAJOR_VERSION=$( echo "${JAVA_VERSION}" | sed -e 's/^\([0-9]*\)\..*$/\1/' )
|
||||
|
||||
Reference in new issue
Block a user