Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/1832#discussion_r100938344
--- Diff: storm-core/src/jvm/org/apache/storm/utils/Time.java ---
@@ -88,59 +92,97 @@ public static boolean isSimulating() {
public static void sleepUntil(long targetTimeMs) throws
InterruptedException {
if(simulating.get()) {
- try {
- synchronized(sleepTimesLock) {
- if (threadSleepTimes == null) {
+ simulatedSleepUntilNanos(millisToNanos(targetTimeMs));
+ } else {
+ long sleepTimeMs = targetTimeMs - currentTimeMillis();
+ if(sleepTimeMs>0) {
+ Thread.sleep(sleepTimeMs);
+ }
+ }
+ }
+
+ public static void sleepUntilNanos(long targetTimeNanos) throws
InterruptedException {
+ if(simulating.get()) {
+ simulatedSleepUntilNanos(targetTimeNanos);
+ } else {
+ long sleepTimeNanos = targetTimeNanos-nanoTime();
+ long sleepTimeMs = nanosToMillis(sleepTimeNanos);
+ int sleepTimeNanosSansMs = (int)(sleepTimeNanos%1_000_000);
+ if(sleepTimeNanos>0) {
+ Thread.sleep(sleepTimeMs, sleepTimeNanosSansMs);
+ }
+ }
+ }
+
+ private static void simulatedSleepUntilNanos(long targetTimeNanos)
throws InterruptedException {
+ try {
+ synchronized (sleepTimesLock) {
+ if (threadSleepTimesNanos == null) {
+ LOG.debug("{} is still sleeping after simulated time
disabled.", Thread.currentThread(), new RuntimeException("STACK TRACE"));
+ throw new InterruptedException();
+ }
+ threadSleepTimesNanos.put(Thread.currentThread(), new
AtomicLong(targetTimeNanos));
+ }
+ while (simulatedCurrTimeNanos.get() < targetTimeNanos) {
+ synchronized (sleepTimesLock) {
+ if (threadSleepTimesNanos == null) {
LOG.debug("{} is still sleeping after simulated
time disabled.", Thread.currentThread(), new RuntimeException("STACK TRACE"));
throw new InterruptedException();
}
- threadSleepTimes.put(Thread.currentThread(), new
AtomicLong(targetTimeMs));
}
- while(simulatedCurrTimeMs.get() < targetTimeMs) {
- synchronized(sleepTimesLock) {
- if (threadSleepTimes == null) {
- LOG.debug("{} is still sleeping after
simulated time disabled.", Thread.currentThread(), new RuntimeException("STACK
TRACE"));
- throw new InterruptedException();
- }
- }
- long autoAdvance = autoAdvanceOnSleep.get();
- if (autoAdvance > 0) {
- advanceTime(autoAdvance);
- }
- Thread.sleep(10);
+ long autoAdvance = autoAdvanceNanosOnSleep.get();
+ if (autoAdvance > 0) {
+ advanceTimeNanos(autoAdvance);
}
- } finally {
- synchronized(sleepTimesLock) {
- if (simulating.get() && threadSleepTimes != null) {
- threadSleepTimes.remove(Thread.currentThread());
- }
+ Thread.sleep(10);
+ }
+ } finally {
+ synchronized (sleepTimesLock) {
+ if (simulating.get() && threadSleepTimesNanos != null) {
+ threadSleepTimesNanos.remove(Thread.currentThread());
}
}
- } else {
- long sleepTime = targetTimeMs-currentTimeMillis();
- if(sleepTime>0)
- Thread.sleep(sleepTime);
}
}
public static void sleep(long ms) throws InterruptedException {
sleepUntil(currentTimeMillis()+ms);
}
+
+ public static void sleepNanos(long nanos) throws InterruptedException {
+ sleepUntilNanos(nanoTime() + nanos);
+ }
public static void sleepSecs (long secs) throws InterruptedException {
if (secs > 0) {
sleep(secs * 1000);
}
}
+ public static long nanoTime() {
+ if (simulating.get()) {
+ return simulatedCurrTimeNanos.get();
+ } else {
+ return System.nanoTime();
+ }
+ }
+
public static long currentTimeMillis() {
if(simulating.get()) {
- return simulatedCurrTimeMs.get();
+ return nanosToMillis(simulatedCurrTimeNanos.get());
--- End diff --
I believe this is not an atomic operation
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
