litvinovg/vivo
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Threading improvements (part 2)

Browse source
This commit is contained in:
Graham Triggs 2016-07-28 22:39:05 +01:00
parent 4d8bfe1a65
commit 2ca805b4ff
1 changed files with 29 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
api/src/main/java/edu/cornell/mannlib/vitro/webapp/visualization/utilities/CachingRDFServiceExecutor.java
View file

@ -11,6 +11,7 @@ import java.util.HashSet;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
import java.util.concurrent.Callable; import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask; import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@ -380,18 +381,19 @@ public class CachingRDFServiceExecutor<T> {
public static class Affinity { public static class Affinity {
private final int maxThreads = 1; private final int maxThreads = 1;
static class ThreadTimers { static class ThreadControl {
ThreadTimers(long started, long expectedDuration) { ThreadControl(long started, long expectedDuration) {
this.started = started; this.started = started;
this.expectedDuration = expectedDuration; this.expectedDuration = expectedDuration;
} }
final long started; final long started;
final long expectedDuration; final long expectedDuration;
final CountDownLatch latch = new CountDownLatch(1);
} }
// Map of executing threads, and the time they expect to need to execute // Map of executing threads, and the time they expect to need to execute
private final Map<Thread, ThreadTimers> threadToExecutionTime = new HashMap<>(); private final Map<Thread, ThreadControl> threadToExecutionTime = new HashMap<>();
private final Set<Thread> executingThreads = new HashSet<>(); private final Set<Thread> executingThreads = new HashSet<>();
/** /**
@ -402,15 +404,13 @@ public class CachingRDFServiceExecutor<T> {
Thread executingThread = Thread.currentThread(); Thread executingThread = Thread.currentThread();
// Ask if the task needs to be queued // Ask if the task needs to be queued
if (queueThis(executingThread, expectedExecutionTime)) { CountDownLatch latch = queueThis(executingThread, expectedExecutionTime);
// Synchronize the thread to call wait
synchronized (executingThread) { // We got a latch from the queue, so wait for it to clear
if (latch != null) {
try { try {
// Make the thread wait until it is notified to continue latch.await();
executingThread.wait();
} catch (InterruptedException e) { } catch (InterruptedException e) {
// No need to execute
}
} }
} }
} }
@ -421,20 +421,22 @@ public class CachingRDFServiceExecutor<T> {
* @param time start time of the thread * @param time start time of the thread
* @return true if the thread needs to wait, false if it can continue * @return true if the thread needs to wait, false if it can continue
*/ */
private synchronized boolean queueThis(Thread thread, Long time) { private synchronized CountDownLatch queueThis(Thread thread, Long time) {
// If we have fewer that the max threads running // If we have fewer that the max threads running
if (executingThreads.size() < maxThreads) { if (executingThreads.size() < maxThreads) {
// Add thread to executing set // Add thread to executing set
executingThreads.add(thread); executingThreads.add(thread);
// Not queued - we can continue // Not queued - we can continue
return false; return null;
} else { } else {
// Add the thread to the map ThreadControl control = new ThreadControl(System.currentTimeMillis(), time);
threadToExecutionTime.put(thread, new ThreadTimers(System.currentTimeMillis(), time));
// Let the caller know that we are queued // Add the thread to the map
return true; threadToExecutionTime.put(thread, control);
// Give the caller a handle to the latch for the queued thread
return control.latch;
} }
} }
@ -451,36 +453,32 @@ public class CachingRDFServiceExecutor<T> {
// If there are still threads to execute, and we have not exhausted maximum threads // If there are still threads to execute, and we have not exhausted maximum threads
while (threadToExecutionTime.size() > 0 && executingThreads.size() < maxThreads) { while (threadToExecutionTime.size() > 0 && executingThreads.size() < maxThreads) {
Thread nextToRelease = null; Thread nextToRelease = null;
long executionTime = -1; ThreadControl nextToReleaseControl = null;
long started = -1;
long current = System.currentTimeMillis(); long current = System.currentTimeMillis();
boolean favourStartTime = false; boolean favourStartTime = false;
// Find the thread that expects to take the least time // Find the thread that expects to take the least time
for (Thread thread : threadToExecutionTime.keySet()) { for (Thread thread : threadToExecutionTime.keySet()) {
ThreadTimers thisThreadTimer = threadToExecutionTime.get(thread); ThreadControl threadControl = threadToExecutionTime.get(thread);
// If there are threads that have been waiting over 2 seconds, favour the oldest thread // If there are threads that have been waiting over 2 seconds, favour the oldest thread
if (thisThreadTimer.started + 2000 < current) { if (threadControl.started + 2000 < current) {
favourStartTime = true; favourStartTime = true;
} }
if (nextToRelease == null) { if (nextToRelease == null) {
nextToRelease = thread; nextToRelease = thread;
executionTime = thisThreadTimer.expectedDuration; nextToReleaseControl = threadControl;
started = thisThreadTimer.started;
} else { } else {
if (favourStartTime) { if (favourStartTime) {
// Find the oldest thread // Find the oldest thread
if (thisThreadTimer.started < started) { if (threadControl.started < nextToReleaseControl.started) {
nextToRelease = thread; nextToRelease = thread;
executionTime = thisThreadTimer.expectedDuration; nextToReleaseControl = threadControl;
started = thisThreadTimer.started;
} }
} else if (thisThreadTimer.expectedDuration < executionTime) { } else if (threadControl.expectedDuration < nextToReleaseControl.expectedDuration) {
nextToRelease = thread; nextToRelease = thread;
executionTime = thisThreadTimer.expectedDuration; nextToReleaseControl = threadControl;
started = thisThreadTimer.started;
} }
} }
} }
@ -489,9 +487,7 @@ public class CachingRDFServiceExecutor<T> {
if (nextToRelease != null) { if (nextToRelease != null) {
threadToExecutionTime.remove(nextToRelease); threadToExecutionTime.remove(nextToRelease);
executingThreads.add(nextToRelease); executingThreads.add(nextToRelease);
synchronized (nextToRelease) { nextToReleaseControl.latch.countDown();
nextToRelease.notifyAll();
}
} }
} }
} }