acquire locks using a timeout so that we don't wait forever for what may be a Java VM bug

See also http://bugs.sun.com/view_bug.do?bug_id=6822370. As an additional means to robustness, added an interruption timer that will interrupt the live leaderboard updater thread if it hangs for more than the update interval
2026-10-21 11:40:01 +00:00 · 2012-07-23 23:56:41 +02:00
1 parent ecb2b8320f
commit e9464b05d4
@@ -4,8 +4,42 @@ import java.io.Serializable;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.logging.Logger;

 public class Util {
+    private static final int NUMBER_OF_SECONDS_TO_WAIT_FOR_LOCK = 5;
+    private static final Logger logger = Logger.getLogger(Util.class.getName());
+    
+    /**
+     * Bug <a href="http://bugs.sun.com/view_bug.do?bug_id=6822370">http://bugs.sun.com/view_bug.do?bug_id=6822370</a> seems
+     * dangerous, particularly if it happens in a <code>LiveLeaderboardUpdater</code> thread. Even though the bug is reported to
+     * have been fixed in JDK 7(b79) we should be careful. This method tries to acquire a lock, allowing for five seconds to pass.
+     * After five seconds and not having retrieved the lock, tries again until the lock has been acquired.
+     * @throws InterruptedException 
+     */
+    public static void lock(Lock lock) {
+        boolean locked = false;
+        boolean interrupted = false;
+        while (!locked) {
+            try {
+                locked = lock.tryLock(NUMBER_OF_SECONDS_TO_WAIT_FOR_LOCK, TimeUnit.SECONDS);
+                if (!locked) {
+                    logger.info("Couldn't acquire lock in "+NUMBER_OF_SECONDS_TO_WAIT_FOR_LOCK+"s. Trying again...");
+                }
+            }
+            catch (InterruptedException ex) {
+                interrupted = true;
+            }
+        }
+        if (interrupted) {
+            // re-assert interrupt state that occurred while we
+            // were acquiring the lock
+            Thread.currentThread().interrupt();
+        }
+    }
+    
    /**
     * Adds all elements from <code>what</code> to <code>addTo</code> and returns <code>addTo</code> for chained use.
     */
@@ -30,7 +64,7 @@ public class Util {
        }
    }
    
-    public static <T> int indexOf(Iterable<? extends T> i, T t) {
+    public static <T> int indexO(Iterable<? extends T> i, T t) {
        int result;
        if (i instanceof List<?>) {
            List<?> list = (List<?>) i;
@@ -24,6 +24,7 @@ import com.sap.sailing.domain.base.DomainFactory;
 import com.sap.sailing.domain.base.Leg;
 import com.sap.sailing.domain.base.Waypoint;
 import com.sap.sailing.domain.common.impl.NamedImpl;
+import com.sap.sailing.domain.common.impl.Util;
 import com.sap.sailing.util.CourseAsWaypointList;

 import difflib.DiffUtils;
@@ -68,7 +69,7 @@ public class CourseImpl extends NamedImpl implements Course {
    
    @Override
    public void lockForRead() {
-        lock.readLock().lock();
+        Util.lock(lock.readLock());
    }

    @Override
@@ -108,7 +109,7 @@ public class CourseImpl extends NamedImpl implements Course {
    
    @Override
    public void addWaypoint(int zeroBasedPosition, Waypoint waypointToAdd) {
-        lock.writeLock().lock();
+        Util.lock(lock.writeLock());
        try {
            waypoints.add(zeroBasedPosition, waypointToAdd);
            Map<Waypoint, Integer> updatesToWaypointIndexes = new HashMap<Waypoint, Integer>();
@@ -138,7 +139,7 @@ public class CourseImpl extends NamedImpl implements Course {
    public void removeWaypoint(int zeroBasedPosition) {
        if (zeroBasedPosition >= 0) {
            Waypoint removedWaypoint;
-            lock.writeLock().lock();
+            Util.lock(lock.writeLock());
            try {
                boolean isLast = zeroBasedPosition == waypoints.size() - 1;
                removedWaypoint = waypoints.remove(zeroBasedPosition);
@@ -353,7 +354,7 @@ public class CourseImpl extends NamedImpl implements Course {

    @Override
    public void update(List<ControlPoint> newControlPoints, DomainFactory baseDomainFactory) throws PatchFailedException {
-        lock.writeLock().lock();
+        Util.lock(lock.writeLock());
        try {
            Iterable<Waypoint> courseWaypoints = getWaypoints();
            List<Waypoint> newWaypointList = new LinkedList<Waypoint>();
@@ -26,6 +26,7 @@ import com.sap.sailing.domain.common.TimePoint;
 import com.sap.sailing.domain.common.WindSource;
 import com.sap.sailing.domain.common.WindSourceType;
 import com.sap.sailing.domain.common.impl.DegreeBearingImpl;
+import com.sap.sailing.domain.common.impl.Util;
 import com.sap.sailing.domain.common.impl.Util.Pair;
 import com.sap.sailing.domain.confidence.ConfidenceFactory;
 import com.sap.sailing.domain.confidence.Weigher;
@@ -182,7 +183,7 @@ public class TrackBasedEstimationWindTrackImpl extends VirtualWindTrackImpl impl
     * Synchronizes serialization on this object to avoid the cache being updated while being written.
     */
    private synchronized void writeObject(ObjectOutputStream s) throws IOException {
-        cacheLock.readLock().lock();
+        Util.lock(cacheLock.readLock());
        lockForRead();
        try {
            s.defaultWriteObject();
@@ -227,7 +228,7 @@ public class TrackBasedEstimationWindTrackImpl extends VirtualWindTrackImpl impl
        // can't use lockForWrite() here because caching can happen while holding the read lock, and the lock can't be
        // upgraded. But lockForRead() and synchronization will do the job because all invalidations lock the write lock,
        // and all contains() checks and get() calls use synchronization too.
-        cacheLock.writeLock().lock();
+        Util.lock(cacheLock.writeLock());
        try {
            // synchronization necessary to protect writeObject from ConcurrentModificationException
            synchronized (this) {
@@ -239,7 +240,7 @@ public class TrackBasedEstimationWindTrackImpl extends VirtualWindTrackImpl impl
                        timePointsWithCachedNullResultFastContains.add(timePoint);
                    }
                } else {
-                    cacheLock.writeLock().lock();
+                    Util.lock(cacheLock.writeLock());
                    try {
                        cache.add(fix);
                    } finally {
@@ -281,7 +282,7 @@ public class TrackBasedEstimationWindTrackImpl extends VirtualWindTrackImpl impl
     * running.
     */
    private void invalidateCache() {
-        cacheLock.writeLock().lock();
+        Util.lock(cacheLock.writeLock());
        try {
            Iterator<WindWithConfidence<TimePoint>> iter = (scheduledRefreshInterval.getStart() == null ? getCachedFixes()
                    : getCachedFixes().tailSet(scheduledRefreshInterval.getStart(), /* inclusive */true)).iterator();
@@ -317,7 +318,7 @@ public class TrackBasedEstimationWindTrackImpl extends VirtualWindTrackImpl impl
    private void refreshCacheIncrementally() {
        Set<WindWithConfidence<TimePoint>> windFixesToRecalculate = new HashSet<WindWithConfidence<TimePoint>>();
        Set<TimePoint> cachedNullResultsToRecalculate = new HashSet<TimePoint>();
-        cacheLock.readLock().lock();
+        Util.lock(cacheLock.readLock());
        try {
            Iterator<WindWithConfidence<TimePoint>> iter = (scheduledRefreshInterval.getStart() == null ? getCachedFixes()
                    : getCachedFixes().tailSet(scheduledRefreshInterval.getStart(), /* inclusive */true)).iterator();
@@ -362,7 +363,7 @@ public class TrackBasedEstimationWindTrackImpl extends VirtualWindTrackImpl impl
            }
        }
        // apply the computed cache deltas
-        cacheLock.writeLock().lock();
+        Util.lock(cacheLock.writeLock());
        try {
            for (TimePoint nullRemoval : nullRemovals) {
                timePointsWithCachedNullResult.remove(nullRemoval);
@@ -402,7 +403,7 @@ public class TrackBasedEstimationWindTrackImpl extends VirtualWindTrackImpl impl
    }

    private void clearCache() {
-        cacheLock.writeLock().lock();
+        Util.lock(cacheLock.writeLock());
        try {
            cache.clear();
            timePointsWithCachedNullResult.clear();
@@ -422,7 +423,7 @@ public class TrackBasedEstimationWindTrackImpl extends VirtualWindTrackImpl impl
        WindWithConfidence<TimePoint> cachedFix = null;
        WindWithConfidence<TimePoint> result = null;
        final boolean nullResultCacheContains;
-        cacheLock.readLock().lock();
+        Util.lock(cacheLock.readLock());
        try {
            nullResultCacheContains = nullResultCacheContains(timePoint);
            if (nullResultCacheContains) {
@@ -9,6 +9,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;

 import com.sap.sailing.domain.base.Timed;
 import com.sap.sailing.domain.common.TimePoint;
+import com.sap.sailing.domain.common.impl.Util;
 import com.sap.sailing.domain.tracking.Track;
 import com.sap.sailing.util.impl.ArrayListNavigableSet;
 import com.sap.sailing.util.impl.UnmodifiableNavigableSet;
@@ -62,7 +63,7 @@ public class TrackImpl<FixType extends Timed> implements Track<FixType> {

    @Override
    public void lockForRead() {
-        readWriteLock.readLock().lock();
+        Util.lock(readWriteLock.readLock());
    }

    @Override
@@ -71,7 +72,7 @@ public class TrackImpl<FixType extends Timed> implements Track<FixType> {
    }
    
    protected void lockForWrite() {
-        readWriteLock.writeLock().lock();
+        Util.lock(readWriteLock.writeLock());
    }
    
    protected void unlockAfterWrite() {
@@ -755,7 +755,7 @@ public abstract class TrackedRaceImpl implements TrackedRace, CourseListener {
                lock = readWriteLock.readLock();
            }
        }
-        lock.lock();
+        Util.lock(lock);
        try {
            if (rankedCompetitors == null) {
                rankedCompetitors = competitorRankings.get(timePoint); // try again; maybe a writer released the write lock after updating the cache
@@ -1353,6 +1353,7 @@ public abstract class TrackedRaceImpl implements TrackedRace, CourseListener {
        for (LegType legType : LegType.values()) {
            bearings.put(legType, new BearingWithConfidenceCluster<TimePoint>(weigher));
        }
+        Map<TrackedLeg, LegType> legTypesCache = new HashMap<TrackedLeg, LegType>();
        getRace().getCourse().lockForRead(); // ensure the course doesn't change, particularly lose the leg we're interested in, while we're running
        try {
            for (Competitor competitor : getRace().getCompetitors()) {
@@ -1361,7 +1362,11 @@ public abstract class TrackedRaceImpl implements TrackedRace, CourseListener {
                    TrackedLeg trackedLeg = getTrackedLeg(leg.getLeg());
                    LegType legType;
                    try {
-                        legType = trackedLeg.getLegType(timePoint);
+                        legType = legTypesCache.get(trackedLeg);
+                        if (legType == null) {
+                            legType = trackedLeg.getLegType(timePoint);
+                            legTypesCache.put(trackedLeg, legType);
+                        }
                        if (legType != LegType.REACHING) {
                            GPSFixTrack<Competitor, GPSFixMoving> track = getTrack(competitor);
                            if (!track.hasDirectionChange(timePoint, getManeuverDegreeAngleThreshold())) {
@@ -12,6 +12,7 @@ import java.util.concurrent.FutureTask;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.logging.Logger;

+import com.sap.sailing.domain.common.impl.Util;
 import com.sap.sailing.util.impl.SmartFutureCache.UpdateInterval;

 /**
@@ -192,7 +193,7 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
                        public V call() {
                            try {
                                V preResult = cacheUpdateComputer.computeCacheUpdate(key, joinedUpdateInterval);
-                                getOrCreateLockForKey(key).writeLock().lock();
+                                Util.lock(getOrCreateLockForKey(key).writeLock());
                                try {
                                    V result = cacheUpdateComputer.provideNewCacheValue(key, cache.get(key), preResult, joinedUpdateInterval);
                                    if (result == null) {
@@ -238,7 +239,7 @@ public class SmartFutureCache<K, V, U extends UpdateInterval<U>> {
                }
            } // else no calculation currently going on; value has been fetched from latest cache entry
        } else {
-            getOrCreateLockForKey(key).readLock().lock();
+            Util.lock(getOrCreateLockForKey(key).readLock());
            try {
                value = cache.get(key);
            } finally {
@@ -5,6 +5,8 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.logging.Logger;

 import com.sap.sailing.domain.base.impl.MillisecondsTimePoint;
@@ -93,6 +95,12 @@ public class LiveLeaderboardUpdater implements Runnable {
    
    private int cacheHitCount;
    private int cacheMissCount;
+
+    /**
+     * If {@link #running}, holds the thread created by {@link #start()}. This thread will be stopped if computing an update
+     * takes longer than {@link #UPDATE_TIMEOUT_IN_MILLIS} milliseconds.
+     */
+    private Thread thread;
    
    public LiveLeaderboardUpdater(String leaderboardName, SailingServiceImpl sailingService) {
        this.leaderboardName = leaderboardName;
@@ -138,7 +146,8 @@ public class LiveLeaderboardUpdater implements Runnable {

    private synchronized void start() {
        running = true;
-        new Thread(this, "LiveLeaderboardUpdater for leaderboard "+getLeaderboard().getName()).start();
+        thread = new Thread(this, "LiveLeaderboardUpdater for leaderboard "+getLeaderboard().getName());
+        thread.start();
    }

    /**
@@ -179,42 +188,69 @@ public class LiveLeaderboardUpdater implements Runnable {
    @Override
    public void run() {
        assert running;
-        logger.info("Starting "+LiveLeaderboardUpdater.class.getSimpleName()+" thread for leaderboard "+leaderboardName);
-        while (true) {
-            MillisecondsTimePoint now = MillisecondsTimePoint.now();
-            TimePoint timePoint = now.minus(getLeaderboard().getDelayToLiveInMillis());
-            synchronized (this) {
-                if (timePoint.asMillis() - lastRequest.asMillis() >= UPDATE_TIMEOUT_IN_MILLIS) {
-                    running = false;
-                    break; // make sure no-one sets running to true again while outside the synchronized block and before re-evaluating the while condition
+        try {
+            logger.info("Starting " + LiveLeaderboardUpdater.class.getSimpleName() + " thread for leaderboard "
+                    + leaderboardName);
+            // interrupt the current thread if not producing a single result within the overall timeout
+            Timer interruptTimer = new Timer("Interrupt timer for "+LiveLeaderboardUpdater.class.getSimpleName()+" for leaderboard "+leaderboardName);
+            while (true) {
+                MillisecondsTimePoint now = MillisecondsTimePoint.now();
+                TimePoint timePoint = now.minus(getLeaderboard().getDelayToLiveInMillis());
+                synchronized (this) {
+                    if (timePoint.asMillis() - lastRequest.asMillis() >= UPDATE_TIMEOUT_IN_MILLIS) {
+                        running = false;
+                        break; // make sure no-one sets running to true again while outside the synchronized block and
+                               // before re-evaluating the while condition
+                    }
                }
-            }
-            TimePoint timeLastUpdateWasStarted = now;
-            try {
-                final Set<String> namesOfRaceColumnsForWhichToLoadLegDetails = getColumnNamesForWhichToFetchDetails(timePoint);
-                LeaderboardDTO newCacheValue = sailingService.computeLeaderboardByName(leaderboardName, timePoint,
-                        namesOfRaceColumnsForWhichToLoadLegDetails, /* waitForLatestAnalyses */ false);
-                updateCacheContents(namesOfRaceColumnsForWhichToLoadLegDetails, newCacheValue);
-            } catch (NoWindException e) {
-                logger.info("Unable to update cached leaderboard results for leaderboard "+leaderboardName+": "+e.getMessage());
-                logger.throwing(LiveLeaderboardUpdater.class.getName(), "run", e);
+                TimePoint timeLastUpdateWasStarted = now;
                try {
-                    Thread.sleep(1000); // avoid running into the same NoWindException too quickly
-                } catch (InterruptedException e1) {
-                    logger.throwing(LiveLeaderboardUpdater.class.getName(), "run", e1);
-                }
-            }
-            TimePoint computeLeaderboardByNameFinishedAt = MillisecondsTimePoint.now();
-            long millisToSleep = MINIMUM_TIME_BETWEEN_UPDATES - (computeLeaderboardByNameFinishedAt.asMillis()-timeLastUpdateWasStarted.asMillis());
-            if (millisToSleep > 0) {
-                try {
-                    Thread.sleep(millisToSleep);
-                } catch (InterruptedException e) {
+                    final Set<String> namesOfRaceColumnsForWhichToLoadLegDetails = getColumnNamesForWhichToFetchDetails(timePoint);
+                    TimerTask interruptTask = new TimerTask() {
+                        @Override
+                        public void run() {
+                            synchronized (LiveLeaderboardUpdater.this) {
+                                if (thread != null) {
+                                    logger.severe("Interrupting "+LiveLeaderboardUpdater.class.getSimpleName()+" thread "+thread+
+                                            " after not receiving an update in "+UPDATE_TIMEOUT_IN_MILLIS+"ms");
+                                    thread.interrupt();
+                                }
+                            }
+                        }
+                    };
+                    interruptTimer.schedule(interruptTask, UPDATE_TIMEOUT_IN_MILLIS);
+                    LeaderboardDTO newCacheValue = sailingService.computeLeaderboardByName(leaderboardName, timePoint,
+                            namesOfRaceColumnsForWhichToLoadLegDetails, /* waitForLatestAnalyses */false);
+                    interruptTask.cancel();
+                    updateCacheContents(namesOfRaceColumnsForWhichToLoadLegDetails, newCacheValue);
+                } catch (NoWindException e) {
+                    logger.info("Unable to update cached leaderboard results for leaderboard " + leaderboardName + ": "
+                            + e.getMessage());
                    logger.throwing(LiveLeaderboardUpdater.class.getName(), "run", e);
+                    try {
+                        Thread.sleep(1000); // avoid running into the same NoWindException too quickly
+                    } catch (InterruptedException e1) {
+                        logger.throwing(LiveLeaderboardUpdater.class.getName(), "run", e1);
+                    }
+                }
+                TimePoint computeLeaderboardByNameFinishedAt = MillisecondsTimePoint.now();
+                long millisToSleep = MINIMUM_TIME_BETWEEN_UPDATES
+                        - (computeLeaderboardByNameFinishedAt.asMillis() - timeLastUpdateWasStarted.asMillis());
+                if (millisToSleep > 0) {
+                    try {
+                        Thread.sleep(millisToSleep);
+                    } catch (InterruptedException e) {
+                        logger.throwing(LiveLeaderboardUpdater.class.getName(), "run", e);
+                    }
                }
            }
+            logger.info("" + LiveLeaderboardUpdater.class.getSimpleName() + " thread for leaderboard "
+                    + leaderboardName + " ending");
+        } catch (Throwable t) {
+            running = false;
+            logger.info("exception in "+LiveLeaderboardUpdater.class.getName()+".run(): "+t.getMessage());
+            logger.throwing(LiveLeaderboardUpdater.class.getName(), "run", t);
        }
-        logger.info(""+LiveLeaderboardUpdater.class.getSimpleName()+" thread for leaderboard "+leaderboardName+" ending");
    }

    /**