| 446 | Operator* op, |
| 447 | const char* operatorMethod) { |
| 448 | return fmt::format("{}::{}", op ? op->operatorType() : "N/A", operatorMethod); |
| 449 | } |
| 450 | |
| 451 | CpuWallTiming Driver::processLazyTiming( |
| 452 | Operator& op, |
| 453 | const CpuWallTiming& timing) { |
| 454 | if (&op == operators_[0].get()) { |
| 455 | return timing; |
| 456 | } |
| 457 | auto lockStats = op.stats().wlock(); |
| 458 | uint64_t cpuDelta = 0; |
| 459 | uint64_t wallDelta = 0; |
| 460 | auto it = lockStats->runtimeStats.find(LazyVector::kCpuNanos); |
| 461 | if (it != lockStats->runtimeStats.end()) { |
| 462 | auto cpu = it->second.sum; |
| 463 | cpuDelta = cpu >= lockStats->lastLazyCpuNanos |
| 464 | ? cpu - lockStats->lastLazyCpuNanos |
| 465 | : 0; |
| 466 | if (cpuDelta == 0) { |
| 467 | // return early if no change. Checking one counter is enough. If |
| 468 | // this did not change and the other did, the change would be |
| 469 | // insignificant and tracking would catch up when this counter next |
| 470 | // changed. |
| 471 | return timing; |
| 472 | } |
| 473 | lockStats->lastLazyCpuNanos = cpu; |
| 474 | } else { |
| 475 | // Return early if no lazy activity. Lazy CPU and wall times are recorded |
| 476 | // together, checking one is enough. |
| 477 | return timing; |
| 478 | } |
| 479 | it = lockStats->runtimeStats.find(LazyVector::kWallNanos); |
| 480 | if (it != lockStats->runtimeStats.end()) { |
| 481 | auto wall = it->second.sum; |
| 482 | wallDelta = wall >= lockStats->lastLazyWallNanos |
| 483 | ? wall - lockStats->lastLazyWallNanos |
| 484 | : 0; |
| 485 | if (wallDelta > 0) { |
| 486 | lockStats->lastLazyWallNanos = wall; |
| 487 | } |
| 488 | } |
| 489 | |
| 490 | // Checks and tries to update input bytes from lazy loads. |
| 491 | int64_t inputBytesDelta = 0; |
| 492 | it = lockStats->runtimeStats.find(LazyVector::kInputBytes); |
| 493 | if (it != lockStats->runtimeStats.end()) { |
| 494 | const int64_t inputBytes = it->second.sum; |
| 495 | inputBytesDelta = inputBytes - lockStats->lastLazyInputBytes; |
| 496 | if (inputBytesDelta > 0) { |
| 497 | lockStats->lastLazyInputBytes = inputBytes; |
| 498 | } |
| 499 | } |
| 500 | |
| 501 | lockStats.unlock(); |
| 502 | cpuDelta = std::min<int64_t>(cpuDelta, timing.cpuNanos); |
| 503 | wallDelta = std::min<int64_t>(wallDelta, timing.wallNanos); |
| 504 | lockStats = operators_[0]->stats().wlock(); |
| 505 | lockStats->getOutputTiming.add(CpuWallTiming{ |
no test coverage detected