Fix incorrect handling of System.nanoTime() in DriverContext
@@ -143,9 +143,10 @@ public class DriverContext
public void startProcessTimer()
{
- if (startNanos.compareAndSet(0, System.nanoTime())) {
- pipelineContext.start();
+ // Must update startNanos first so that the value is valid once executionStartTime is not null
+ if (executionStartTime.get() == null && startNanos.compareAndSet(0, System.nanoTime())) {
executionStartTime.set(DateTime.now());
+ pipelineContext.start();
}
}
@@ -174,8 +175,9 @@ public class DriverContext
// already finished
return;
}
- executionEndTime.set(DateTime.now());
+ // Must update endNanos first, so that the value is valid after executionEndTime is not null
endNanos.set(System.nanoTime());
+ executionEndTime.set(DateTime.now());
pipelineContext.driverFinished(this);
}
@@ -323,6 +325,14 @@ public class DriverContext
totalBlockedTime += blockedMonitor.getBlockedTime();
}
+ // startNanos is always valid once executionStartTime is not null
+ DateTime executionStartTime = this.executionStartTime.get();
+ Duration queuedTime = new Duration(nanosBetween(createNanos, executionStartTime == null ? System.nanoTime() : startNanos.get()), NANOSECONDS);
+
+ // endNanos is always valid once executionStartTime is not null
+ DateTime executionEndTime = this.executionEndTime.get();
+ Duration elapsedTime = new Duration(nanosBetween(createNanos, executionEndTime == null ? System.nanoTime() : endNanos.get()), NANOSECONDS);
+
List<OperatorStats> operators = ImmutableList.copyOf(transform(operatorContexts, OperatorContext::getOperatorStats));
OperatorStats inputOperator = getFirst(operators, null);
DataSize rawInputDataSize;
@@ -356,23 +366,8 @@ public class DriverContext
outputPositions = 0;
}
- long startNanos = this.startNanos.get();
- if (startNanos < createNanos) {
- startNanos = System.nanoTime();
- }
- Duration queuedTime = new Duration(startNanos - createNanos, NANOSECONDS);
-
- long endNanos = this.endNanos.get();
- Duration elapsedTime;
- if (endNanos >= startNanos) {
- elapsedTime = new Duration(endNanos - createNanos, NANOSECONDS);
- }
- else {
- elapsedTime = new Duration(0, NANOSECONDS);
- }
-
- long physicalWrittenDataSize = 0;
ImmutableSet.Builder<BlockedReason> builder = ImmutableSet.builder();
+ long physicalWrittenDataSize = 0;
for (OperatorStats operator : operators) {
physicalWrittenDataSize += operator.getPhysicalWrittenDataSize().toBytes();
if (operator.getBlockedReason().isPresent()) {
@@ -384,8 +379,8 @@ public class DriverContext
return new DriverStats(
lifespan,
createdTime,
- executionStartTime.get(),
- executionEndTime.get(),
+ executionStartTime,
+ executionEndTime,
queuedTime.convertToMostSuccinctTimeUnit(),
elapsedTime.convertToMostSuccinctTimeUnit(),
succinctBytes(driverMemoryContext.getUserMemory()),
