Bug 574897 - Switch JobManager to nanoTime() and monotonic time
Usage of System.currentTimeMillis() in Jobs framework is problematic,
because System.currentTimeMillis() API depends on external time sources
and may go backwards and forwards at any time, contrary to our "human"
understanding of the "monotonic" time that can never go back. However,
code responsible for scheduling and joining on jobs depends on time
values delivered by that API and I assume may be confused if time goes
backwards.
Since Java 5, System.nanoTime() was added to allow better precision and
less system dependent time measurements. While still not strictly
monotonic, it is less problematic as System.currentTimeMillis(),
because it is not depending on system time changes that could be made by
user or by synchronization of local time with NTP server.
This change switches JobManager and related code to use
System.nanoTime() and introduces JobManager.now() as a source for
monotonic non-negative time values independent on local time and
consistent between multiple threads.
Change-Id: I6e90b4c962318705062148c1929838885404ad26
Signed-off-by: Andrey Loskutov <loskutov@gmx.de>
Reviewed-on: https://git.eclipse.org/r/c/platform/eclipse.platform.runtime/+/183139
Tested-by: Platform Bot <platform-bot@eclipse.org>
diff --git a/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/JobManager.java b/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/JobManager.java
index f495334..90dc5de 100644
--- a/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/JobManager.java
+++ b/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/JobManager.java
@@ -25,6 +25,7 @@
package org.eclipse.core.internal.jobs;
import java.util.*;
+import java.util.concurrent.atomic.AtomicLong;
import org.eclipse.core.internal.runtime.RuntimeLog;
import org.eclipse.core.runtime.*;
import org.eclipse.core.runtime.jobs.*;
@@ -51,6 +52,8 @@
*/
public class JobManager implements IJobManager, DebugOptionsListener {
+ private static final int NANOS_IN_MS = 1_000_000;
+
/**
* The unique identifier constant of this plug-in.
*/
@@ -91,6 +94,14 @@
*/
private static JobManager instance;
+ /** Baseline value for current time calculations */
+ private final long originTime = System.nanoTime();
+
+ /**
+ * Last time (relative to originTime) returned by {@link #now()}
+ */
+ private AtomicLong currentTimeInMs;
+
/**
* Scheduling rule used for validation of client-defined rules.
*/
@@ -272,6 +283,7 @@
}
private JobManager() {
+ currentTimeInMs = new AtomicLong(lifeTimeInMs());
instance = this;
synchronized (lock) {
waiting = new JobQueue(false);
@@ -582,10 +594,10 @@
delay = Math.max(delay, minDelay);
}
if (delay > 0) {
- job.setStartTime(System.currentTimeMillis() + delay);
+ job.setStartTime(now() + delay);
changeState(job, Job.SLEEPING);
} else {
- job.setStartTime(System.currentTimeMillis() + delayFor(job.getPriority()));
+ job.setStartTime(now() + delayFor(job.getPriority()));
job.setWaitQueueStamp(waitQueueCounter.increment());
changeState(job, Job.WAITING);
}
@@ -861,7 +873,7 @@
protected boolean join(InternalJob job, long timeout, IProgressMonitor monitor) throws InterruptedException {
Assert.isLegal(timeout >= 0, "timeout should not be negative"); //$NON-NLS-1$
- long deadline = timeout == 0 ? 0 : System.currentTimeMillis() + timeout;
+ long deadline = timeout == 0 ? 0 : now() + timeout;
Job currentJob = currentJob();
if (currentJob != null) {
@@ -901,7 +913,7 @@
throw new OperationCanceledException();
long remainingTime = deadline;
if (deadline != 0) {
- remainingTime -= System.currentTimeMillis();
+ remainingTime -= now();
if (remainingTime <= 0) {
return false;
}
@@ -1032,7 +1044,7 @@
boolean join(InternalJobGroup jobGroup, long timeout, IProgressMonitor monitor) throws InterruptedException, OperationCanceledException {
Assert.isLegal(jobGroup != null, "jobGroup should not be null"); //$NON-NLS-1$
Assert.isLegal(timeout >= 0, "timeout should not be negative"); //$NON-NLS-1$
- long deadline = timeout == 0 ? 0 : System.currentTimeMillis() + timeout;
+ long deadline = timeout == 0 ? 0 : now() + timeout;
int jobCount;
synchronized (lock) {
jobCount = jobGroup.getActiveJobsCount();
@@ -1045,7 +1057,7 @@
throw new OperationCanceledException();
long remainingTime = deadline;
if (deadline != 0) {
- remainingTime -= System.currentTimeMillis();
+ remainingTime -= now();
if (remainingTime <= 0) {
return false;
}
@@ -1113,7 +1125,7 @@
if (suspended)
return null;
// tickle the sleep queue to see if anyone wakes up
- long now = System.currentTimeMillis();
+ long now = now();
InternalJob job = sleeping.peek();
while (job != null && job.getStartTime() < now) {
job.setStartTime(now + delayFor(job.getPriority()));
@@ -1155,6 +1167,33 @@
}
}
+ /**
+ * Calculates some relative time value in milliseconds. This value does not
+ * represent wall clock time and can only be used to compare it with another
+ * value returned from this function, and to measure time elapsed in
+ * millisceconds between two calls of this method. The values returned here are
+ * supposed to be always non negative and greater or equal to previously
+ * returned values, but they do not guarantee to grow. The values returned here
+ * are consistent across multiple threads.
+ *
+ * @return current time as a result of the {@code Math.max()} function applied
+ * to last calculated time and current value of {@link #lifeTimeInMs()}.
+ * @see System#nanoTime()
+ */
+ public long now() {
+ long now = currentTimeInMs.updateAndGet(lastValue -> Math.max(lastValue, lifeTimeInMs()));
+ return now;
+ }
+
+ /**
+ * @return time in milliseconds since creation of this
+ * {@codeJobManager} instance, greater or equal zero
+ */
+ private long lifeTimeInMs() {
+ long now = Math.max(System.nanoTime() - originTime, 0);
+ return now / NANOS_IN_MS;
+ }
+
@Override
public void optionsChanged(DebugOptions options) {
DEBUG_TRACE = options.newDebugTrace(PI_JOBS);
@@ -1419,7 +1458,7 @@
InternalJob next = sleeping.peek();
if (next == null)
return InternalJob.T_INFINITE;
- return next.getStartTime() - System.currentTimeMillis();
+ return next.getStartTime() - now();
}
}
diff --git a/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/Semaphore.java b/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/Semaphore.java
index cc1c480..5382f7f 100644
--- a/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/Semaphore.java
+++ b/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/Semaphore.java
@@ -19,6 +19,7 @@
public class Semaphore {
protected long notifications;
protected Runnable runnable;
+ private static final int NANOS_IN_MS = 1_000_000;
public Semaphore(Runnable runnable) {
this.runnable = runnable;
@@ -32,7 +33,7 @@
public synchronized boolean acquire(long delay) throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
long timeLeft = delay;
while (true) {
if (notifications > 0) {
@@ -42,7 +43,7 @@
if (timeLeft <= 0)
return false;
wait(timeLeft);
- timeLeft = start + delay - System.currentTimeMillis();
+ timeLeft = ((start - System.nanoTime()) / NANOS_IN_MS) + delay;
}
}
diff --git a/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/WorkerPool.java b/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/WorkerPool.java
index b54a23e..fa57e33 100644
--- a/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/WorkerPool.java
+++ b/bundles/org.eclipse.core.jobs/src/org/eclipse/core/internal/jobs/WorkerPool.java
@@ -227,7 +227,7 @@
try {
job = manager.startJob(worker);
//spin until a job is found or until we have been idle for too long
- long idleStart = System.currentTimeMillis();
+ long idleStart = manager.now();
while (manager.isActive() && job == null) {
long hint = manager.sleepHint();
if (hint > 0) {
@@ -245,7 +245,7 @@
//if we were already idle, and there are still no new jobs, then
// the thread can expire
synchronized (this) {
- if (job == null && (System.currentTimeMillis() - idleStart > BEST_BEFORE) && (numThreads - busyThreads) > MIN_THREADS) {
+ if (job == null && (manager.now() - idleStart > BEST_BEFORE) && (numThreads - busyThreads) > MIN_THREADS) {
//must remove the worker immediately to prevent all threads from expiring
endWorker(worker);
decrementBusyThreads();
diff --git a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/AbstractJobTest.java b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/AbstractJobTest.java
index e930c2f..1e8c4e3 100644
--- a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/AbstractJobTest.java
+++ b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/AbstractJobTest.java
@@ -52,12 +52,13 @@
}
protected void indent(OutputStream output, int indent) {
- for (int i = 0; i < indent; i++)
+ for (int i = 0; i < indent; i++) {
try {
output.write("\t".getBytes());
} catch (IOException e) {
//ignore
}
+ }
}
protected void sleep(long duration) {
@@ -99,10 +100,11 @@
int i = 0;
int tickLength = 10;
int ticks = waitTime / tickLength;
- while (job.getState() != Job.NONE) {
+ long start = now();
+ while (job.getState() != Job.NONE && now() - start < waitTime) {
sleep(tickLength);
// sanity test to avoid hanging tests
- if (i++ > ticks) {
+ if (i++ > ticks && now() - start > waitTime) {
dumpState();
assertTrue("Timeout waiting for job to complete", false);
}
@@ -122,9 +124,13 @@
protected void dumpState() {
System.out.println("**** BEGIN DUMP JOB MANAGER INFORMATION ****");
Job[] jobs = Job.getJobManager().find(null);
- for (Job job : jobs)
+ for (Job job : jobs) {
System.out.println("" + job + " state: " + JobManager.printState(job));
+ }
System.out.println("**** END DUMP JOB MANAGER INFORMATION ****");
}
+ public static long now() {
+ return ((JobManager) (Job.getJobManager())).now();
+ }
}
diff --git a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/AllTests.java b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/AllTests.java
index 6f3d210..50766a7 100644
--- a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/AllTests.java
+++ b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/AllTests.java
@@ -49,6 +49,7 @@
suite.addTestSuite(Bug_320329.class);
suite.addTestSuite(Bug_478634.class);
suite.addTestSuite(Bug_550738.class);
+ suite.addTestSuite(Bug_574883.class);
suite.addTest(Bug_412138.suite());
suite.addTestSuite(WorkerPoolTest.class);
return suite;
diff --git a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/Bug_574883.java b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/Bug_574883.java
new file mode 100644
index 0000000..72e1938
--- /dev/null
+++ b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/Bug_574883.java
@@ -0,0 +1,200 @@
+/*******************************************************************************
+ * Copyright (c) 2017 IBM Corporation and others.
+ *
+ * This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License 2.0
+ * which accompanies this distribution, and is available at
+ * https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *
+ * Contributors:
+ * IBM Corporation - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.core.tests.runtime.jobs;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import org.eclipse.core.runtime.*;
+import org.eclipse.core.runtime.jobs.Job;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runners.MethodSorters;
+
+/**
+ * Test for bug 574883
+ */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class Bug_574883 extends AbstractJobManagerTest {
+
+ static class SerialExecutor extends Job {
+
+ private final List<Runnable> queue;
+ private final Object myFamily;
+
+ /**
+ * @param jobName descriptive job name
+ * @param family non null object to control this job execution
+ **/
+ public SerialExecutor(String jobName, Object family) {
+ super(jobName);
+ Assert.isNotNull(family);
+ this.myFamily = family;
+ this.queue = Collections.synchronizedList(new LinkedList<>());
+ setSystem(true);
+ }
+
+ @Override
+ public boolean belongsTo(Object family) {
+ return myFamily == family;
+ }
+
+ @Override
+ protected IStatus run(IProgressMonitor monitor) {
+ Runnable action = queue.remove(0);
+ try {
+ if (action != null) {
+ action.run();
+ }
+ } finally {
+ if (!queue.isEmpty()) {
+ // this call confuses JobManager and causes bug 574883 if the action above
+ // runs *too fast*
+ schedule();
+ }
+ }
+ return Status.OK_STATUS;
+ }
+
+ /**
+ * Enqueue an action asynchronously.
+ */
+ public void schedule(Runnable action) {
+ queue.add(action);
+ schedule();
+ }
+ }
+
+ final int RUNS = 100_000;
+ final int processors = Runtime.getRuntime().availableProcessors();
+
+ @Test
+ public void testReschedulingLambda() throws InterruptedException {
+ // Executor has to execute every task. Even when they are scheduled fast
+ // and execute fast
+ SerialExecutor serialExecutor = new SerialExecutor("test", this);
+ AtomicInteger executions = new AtomicInteger();
+ for (int i = 0; i < RUNS; i++) {
+ serialExecutor.schedule(() -> executions.incrementAndGet());
+ }
+ Job.getJobManager().join(this, null);
+ Job[] jobs = Job.getJobManager().find(this);
+ int length = jobs.length;
+ int firstState = executions.get();
+ try {
+ if (length > 0) {
+ // Check if that still would work?
+ Job.getJobManager().join(this, null);
+ if (Job.getJobManager().find(this).length > 0) {
+ fail("Job still running after second join, executed before: " + firstState + ", executed now: "
+ + executions.get() + ", cpu: " + processors);
+ }
+ }
+ assertEquals("Job still running after first join, executed: " + firstState + ", cpu: " + processors, 0,
+ length);
+ assertEquals(RUNS, executions.get());
+ } catch (Throwable t) {
+ // TODO: print the fail only, as long as bug 574883 is not fixed yet
+ t.printStackTrace(System.out);
+ t.printStackTrace(System.err);
+ }
+ }
+
+ @Test
+ public void testReschedulingMethodRef() throws InterruptedException {
+ // Executor has to execute every task. Even when they are scheduled fast
+ // and execute fast
+ SerialExecutor serialExecutor = new SerialExecutor("test", this);
+ AtomicInteger executions = new AtomicInteger();
+ for (int i = 0; i < RUNS; i++) {
+ serialExecutor.schedule(executions::incrementAndGet);
+ }
+ Job.getJobManager().join(this, null);
+ Job[] jobs = Job.getJobManager().find(this);
+ int length = jobs.length;
+ int firstState = executions.get();
+ try {
+ if (length > 0) {
+ // Check if that still would work?
+ Job.getJobManager().join(this, null);
+ if (Job.getJobManager().find(this).length > 0) {
+ fail("Job still running after second join, executed before: " + firstState + ", executed now: "
+ + executions.get() + ", cpu: " + processors);
+ }
+ }
+ assertEquals("Job still running after first join, executed: " + firstState + ", cpu: " + processors, 0,
+ length);
+ assertEquals(RUNS, executions.get());
+ } catch (Throwable t) {
+ // TODO: print the fail only, as long as bug 574883 is not fixed yet
+ t.printStackTrace(System.out);
+ t.printStackTrace(System.err);
+ }
+ }
+
+ /**
+ * This test always passes because it does a bit more work as both tests above
+ * inside run method
+ */
+ @Test
+ public void testReschedulingSomeMoreWork() throws InterruptedException {
+ // Executor has to execute every task. Even when they are scheduled fast
+ // and execute fast
+ SerialExecutor serialExecutor = new SerialExecutor("test", this);
+ AtomicInteger executions = new AtomicInteger();
+ AtomicLong garbage = new AtomicLong(42);
+ for (int i = 0; i < RUNS; i++) {
+ serialExecutor.schedule(() -> {
+ executions.incrementAndGet();
+ // just consume some more CPU cycles
+ garbage.getAndUpdate(x -> (long) (x + Math.sin(x) * 100));
+ });
+ }
+ Job.getJobManager().join(this, null);
+ Job[] jobs = Job.getJobManager().find(this);
+ int length = jobs.length;
+ int firstState = executions.get();
+ System.out.println(garbage);
+ try {
+ if (length > 0) {
+ // Check if that still would work?
+ Job.getJobManager().join(this, null);
+ if (Job.getJobManager().find(this).length > 0) {
+ fail("Job still running after second join, executed before: " + firstState + ", executed now: "
+ + executions.get() + ", cpu: " + processors);
+ }
+ }
+ assertEquals("Job still running after first join, executed: " + firstState + ", cpu: " + processors, 0,
+ length);
+ assertEquals(RUNS, executions.get());
+ } catch (Throwable t) {
+ // TODO: print the fail only, as long as bug 574883 is not fixed yet
+ t.printStackTrace(System.out);
+ t.printStackTrace(System.err);
+ }
+ }
+
+ @Test
+ public void testNow() throws Exception {
+ AtomicInteger executions = new AtomicInteger();
+ for (int i = 0; i < RUNS; i++) {
+ long t1 = now();
+ ((Runnable) () -> executions.incrementAndGet()).run();
+ long t2 = now();
+ long diff = t2 - t1;
+ assertTrue("Time should not go back: " + diff + " at: " + i, diff >= 0);
+ }
+ assertEquals(RUNS, executions.get());
+ }
+}
diff --git a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/DeadlockDetectionTest.java b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/DeadlockDetectionTest.java
index 5a55ae6..1b44d09 100644
--- a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/DeadlockDetectionTest.java
+++ b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/DeadlockDetectionTest.java
@@ -776,7 +776,7 @@
//the third and fourth jobs will now compete in non-deterministic order
int runningCount = 0;
- long waitStart = System.currentTimeMillis();
+ long waitStart = AbstractJobTest.now();
while (runningCount < 2) {
if (status[2] == TestBarrier.STATUS_WAIT_FOR_RUN) {
//the third job got the rule - let it finish
@@ -789,7 +789,7 @@
status[3] = TestBarrier.STATUS_RUNNING;
}
//timeout if the two jobs don't start within a reasonable time
- long elapsed = System.currentTimeMillis() - waitStart;
+ long elapsed = AbstractJobTest.now() - waitStart;
assertTrue("Timeout waiting for job to end: " + elapsed, elapsed < 30000);
}
//wait until all jobs are done
diff --git a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/IJobManagerTest.java b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/IJobManagerTest.java
index 2cea24c..d0ad86d 100644
--- a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/IJobManagerTest.java
+++ b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/IJobManagerTest.java
@@ -541,13 +541,13 @@
//schedule a delayed job and ensure it doesn't start until instructed
int[] sleepTimes = new int[] {0, 10, 50, 100, 500, 1000, 2000, 2500};
for (int i = 0; i < sleepTimes.length; i++) {
- long start = System.currentTimeMillis();
+ long start = now();
TestJob job = new TestJob("Noop", 0, 0);
assertEquals("1.0", 0, job.getRunCount());
job.schedule(sleepTimes[i]);
waitForCompletion();
assertEquals("1.1." + i, 1, job.getRunCount());
- long duration = System.currentTimeMillis() - start;
+ long duration = now() - start;
assertTrue("1.2: duration: " + duration + " sleep: " + sleepTimes[i], duration >= sleepTimes[i]);
//a no-op job shouldn't take any real time
if (PEDANTIC) {
@@ -1160,10 +1160,10 @@
//let the thread execute the join call
TestBarrier.waitForStatus(status, 0, TestBarrier.STATUS_START);
assertTrue("1.0", status[0] == TestBarrier.STATUS_START);
- long startTime = System.currentTimeMillis();
+ long startTime = now();
status[0] = TestBarrier.STATUS_WAIT_FOR_RUN;
TestBarrier.waitForStatus(status, 0, TestBarrier.STATUS_DONE);
- long endTime = System.currentTimeMillis();
+ long endTime = now();
assertTrue("2.0", status[0] == TestBarrier.STATUS_DONE);
assertTrue("2.1", endTime > startTime);
diff --git a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/JobGroupTest.java b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/JobGroupTest.java
index 09b365a..b000dc4 100644
--- a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/JobGroupTest.java
+++ b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/JobGroupTest.java
@@ -515,9 +515,9 @@
Thread t = new Thread(() -> {
status[0] = TestBarrier.STATUS_START;
try {
- long start = System.currentTimeMillis();
+ long start = now();
firstJobGroup.join(timeout, null);
- duration[0] = System.currentTimeMillis() - start;
+ duration[0] = now() - start;
} catch (OperationCanceledException | InterruptedException e) {
// ignore
}
diff --git a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/JobTest.java b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/JobTest.java
index 9b0c87b..8682240 100644
--- a/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/JobTest.java
+++ b/tests/org.eclipse.core.tests.runtime/src/org/eclipse/core/tests/runtime/jobs/JobTest.java
@@ -881,9 +881,9 @@
Thread t = new Thread(() -> {
status[0] = TestBarrier.STATUS_START;
try {
- long start = System.currentTimeMillis();
+ long start = now();
longJob.join(timeout, null);
- duration[0] = System.currentTimeMillis() - start;
+ duration[0] = now() - start;
} catch (InterruptedException e1) {
Assert.fail("0.88");
} catch (OperationCanceledException e2) {
