1use crate::{
27 profile::{
28 Profiler, SourceSite, Span, SpanInner, SpanKind, SpanRecord, Trace, advanced, checked,
29 sorted_indices,
30 },
31 sync::Mutex,
32};
33use alloc::{boxed::Box, format, string::String, sync::Arc, vec, vec::Vec};
34use core::{
35 fmt,
36 sync::atomic::{AtomicBool, Ordering},
37 time::Duration,
38};
39use how_far::{
40 Child, ChildPulse, Execution, Outcome, PhaseSpec, PlanError, Pulse, Report, Stop, StopReason,
41};
42use how_far_along::{NodeId, Observer, PulseTree, Snapshot, Status};
43
44#[derive(Clone, Debug)]
46#[non_exhaustive]
47pub struct Options {
48 pub stop_gap_target: Duration,
50 pub report_gap_target: Duration,
52 pub callback_budget: Duration,
54 pub callback_interval_target: Duration,
56 pub check_calls_per_second: f64,
58 pub report_calls_per_second: f64,
60 pub minimum_calls: u64,
62 pub minimum_stage_wall: Duration,
65 pub weight_difference: f64,
68 pub negligible_stage_share: f64,
73 pub uneven_pace_ratio: f64,
79}
80impl Default for Options {
81 fn default() -> Self {
82 Self {
83 stop_gap_target: Duration::from_millis(10),
84 report_gap_target: Duration::from_millis(50),
85 callback_budget: Duration::from_millis(10),
86 callback_interval_target: Duration::from_millis(10),
87 check_calls_per_second: 100_000.0,
88 report_calls_per_second: 10_000.0,
89 minimum_calls: 100,
90 minimum_stage_wall: Duration::from_millis(30),
91 weight_difference: 0.15,
92 negligible_stage_share: 0.02,
93 uneven_pace_ratio: 4.0,
94 }
95 }
96}
97
98#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
100#[non_exhaustive]
101pub enum Kind {
102 AbandonedPhase,
104 InferredCompletion,
106 RecoveredFailure,
108 CountMismatch,
110 ProtocolMisuse,
112 InvalidReport,
114 StopOutcomeMismatch,
116 UnobservedCancellation,
118 StopGap,
120 ReportGap,
122 CheckFrequency,
124 ReportFrequency,
126 CallbackDuration,
128 CallbackInterval,
130 StageWeights,
132 UnevenPace,
135 SuggestedStages,
140 IncompleteEvidence,
142}
143
144#[derive(Clone, Copy, Debug, PartialEq, Eq)]
146#[non_exhaustive]
147pub enum Problem {
148 Plan(PlanError),
150 ReportAfterCompletion,
152 ReportToBranch,
154}
155
156#[derive(Clone, Debug)]
158#[non_exhaustive]
159pub struct Incident {
160 pub node: NodeId,
162 pub problem: Problem,
164 pub site: SourceSite,
166}
167
168#[derive(Clone, Debug)]
172#[non_exhaustive]
173pub struct Finding {
174 pub kind: Kind,
176 pub evidence: String,
178 pub advice: String,
180 pub sample_code: Option<String>,
182}
183impl fmt::Display for Finding {
184 fn fmt(&self, out: &mut fmt::Formatter<'_>) -> fmt::Result {
185 writeln!(out, "{:?}: {}", self.kind, self.evidence)?;
186 writeln!(out, " {}", self.advice)?;
187 if let Some(code) = &self.sample_code {
188 writeln!(out, "{code}")?;
189 }
190 Ok(())
191 }
192}
193
194fn finding(kind: Kind, evidence: String, advice: &str) -> Finding {
195 Finding {
196 kind,
197 evidence,
198 advice: advice.into(),
199 sample_code: None,
200 }
201}
202fn ms(duration: Duration) -> f64 {
203 duration.as_secs_f64() * 1_000.0
204}
205fn location(site: Option<SourceSite>, boundary: &str) -> String {
206 match site {
207 Some(site) => format!("{site}"),
208 None => boundary.into(),
209 }
210}
211
212#[must_use = "finish the measured root after its work joins; dropping it unfinished records Abandoned"]
254pub struct DiagnosticPulse {
255 inner: PulseTree,
256 meter: MeterOwner,
257}
258
259impl fmt::Debug for DiagnosticPulse {
260 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
261 f.debug_struct("DiagnosticPulse")
262 .field("tree", &self.inner)
263 .finish_non_exhaustive()
264 }
265}
266
267impl DiagnosticPulse {
268 pub fn new(tree: PulseTree, profiler: &Profiler) -> Self {
270 profiler.set_report_timing(true);
271 let observer = tree.observer();
272 let name = observer.name().into();
273 Self {
274 meter: Meter::owner(observer, name, profiler, None, None),
275 inner: tree,
276 }
277 }
278
279 pub fn observer(&self) -> Observer {
281 self.meter.observer.clone()
282 }
283
284 pub fn finish(self, outcome: Outcome) -> Result<(), PlanError> {
286 let Self { inner, meter } = self;
287 let result = inner.finish(outcome);
288 meter.close(if result.is_ok() {
289 outcome
290 } else {
291 Outcome::Abandoned
292 });
293 result
294 }
295}
296
297struct MeterOwner(Arc<Meter>);
299impl core::ops::Deref for MeterOwner {
300 type Target = Arc<Meter>;
301 fn deref(&self) -> &Arc<Meter> {
302 &self.0
303 }
304}
305impl Drop for MeterOwner {
306 fn drop(&mut self) {
307 self.close(Outcome::Abandoned);
308 }
309}
310struct Meter {
311 observer: Observer,
313 profiler: Profiler,
314 name: String,
315 span: Mutex<SpanSlot>,
316 entered: Option<Arc<Mutex<Duration>>>,
318 exited: Option<Arc<Mutex<Duration>>>,
320 has_children: AtomicBool,
322 closed: AtomicBool,
323}
324
325#[derive(Default)]
327struct SpanSlot {
328 open: Option<Span>,
329 last: Option<Arc<SpanInner>>,
332}
333impl SpanSlot {
334 fn retire(&mut self, outcome: Outcome) {
336 if let Some(span) = self.open.take() {
337 self.last = Some(span.shared());
338 span.finish(outcome);
339 }
340 }
341}
342
343impl Meter {
344 fn owner(
345 observer: Observer,
346 name: String,
347 profiler: &Profiler,
348 entered: Option<Arc<Mutex<Duration>>>,
349 exited: Option<Arc<Mutex<Duration>>>,
350 ) -> MeterOwner {
351 MeterOwner(Arc::new(Meter {
352 observer,
353 profiler: profiler.clone(),
354 name,
355 span: Mutex::new(SpanSlot::default()),
356 entered,
357 exited,
358 has_children: AtomicBool::new(false),
359 closed: AtomicBool::new(false),
360 }))
361 }
362 #[track_caller]
364 fn check(&self, inner: &dyn Stop) -> Result<(), StopReason> {
365 match self.span() {
366 Some(span) => checked(&span, inner),
367 None => inner.check(),
368 }
369 }
370 #[track_caller]
372 fn advance(&self, inner: &dyn Report, completed: u64) {
373 self.reporting(completed);
374 match self.span() {
375 Some(span) => advanced(&span, inner, completed),
376 None => inner.advance(completed),
377 }
378 }
379 #[track_caller]
380 fn issue(&self, problem: Problem) {
381 self.profiler.incident(Incident {
382 node: self.observer.id(),
383 problem,
384 site: SourceSite::from_location(core::panic::Location::caller()),
385 });
386 }
387 #[track_caller]
388 fn reporting(&self, n: u64) {
389 if n == 0 {
390 return;
391 }
392 if self.closed.load(Ordering::Acquire) || self.observer.is_finished() {
393 self.issue(Problem::ReportAfterCompletion);
394 } else if self.has_children.load(Ordering::Relaxed) {
395 self.issue(Problem::ReportToBranch);
396 }
397 }
398 fn start_span(&self, kind: SpanKind, entered: Option<Duration>) -> Span {
399 let node = Some(self.observer.id());
400 let name = self.name.clone();
401 self.profiler.start_span(None, node, name, kind, entered)
402 }
403 fn start_work(&self) -> Span {
405 let entered = self.entered.as_ref().map(|cell| *cell.lock());
406 self.start_span(SpanKind::Work, entered)
407 }
408
409 fn span(&self) -> Option<Arc<SpanInner>> {
412 let mut slot = self.span.lock();
413 if let Some(span) = &slot.open {
414 return Some(span.shared());
415 }
416 if self.closed.load(Ordering::Acquire) {
417 return slot.last.clone();
418 }
419 let span = self.start_work();
420 let shared = span.shared();
421 slot.open = Some(span);
422 Some(shared)
423 }
424
425 fn delegate(&self) {
431 let mut slot = self.span.lock();
432 if self.closed.load(Ordering::Acquire) || self.has_children.swap(true, Ordering::Relaxed) {
433 return;
434 }
435 let Some(entered) = &self.entered else {
436 slot.retire(Outcome::Succeeded);
437 return;
438 };
439 let now = self.profiler.now();
440 if slot.open.is_none() && *entered.lock() < now {
442 slot.open = Some(self.start_work());
443 }
444 slot.retire(Outcome::Succeeded);
445 slot.open = Some(self.start_span(SpanKind::Wait, Some(now)));
446 }
447
448 fn split<'s>(
449 &self,
450 inner: &'s dyn Pulse,
451 execution: Execution,
452 parts: &[PhaseSpec<'_>],
453 ) -> Result<Vec<Child<'s>>, PlanError> {
454 let children = inner.split(execution, parts)?;
455 self.delegate();
456 let nodes = self.observer.children();
457 let mut cells = Vec::new();
459 if execution == Execution::Sequence {
460 let planned = self.profiler.now();
461 for _ in 0..=children.len() {
462 cells.push(Arc::new(Mutex::new(planned)));
463 }
464 }
465 let mut measured = Vec::with_capacity(children.len());
466 for (index, child) in children.into_iter().enumerate() {
467 let observer = nodes.get(index).unwrap_or(&self.observer).clone();
469 let meter = Meter::owner(
470 observer,
471 parts[index].name.into(),
472 &self.profiler,
473 cells.get(index).cloned(),
474 cells.get(index + 1).cloned(),
475 );
476 measured.push(Child::new(DiagnosticChild {
477 inner: child,
478 meter,
479 }));
480 }
481 Ok(measured)
482 }
483
484 fn close(&self, outcome: Outcome) {
485 if self.closed.swap(true, Ordering::AcqRel) {
486 return;
487 }
488 let mut slot = self.span.lock();
489 if slot.open.is_none()
490 && outcome == Outcome::Succeeded
491 && self.entered.is_some()
492 && !self.has_children.load(Ordering::Relaxed)
493 {
494 slot.open = Some(self.start_work());
496 }
497 slot.retire(outcome);
498 drop(slot);
499 if let Some(exited) = &self.exited {
500 *exited.lock() = self.profiler.now();
502 }
503 }
504}
505
506fn find_node(snapshot: &Snapshot, id: NodeId) -> Option<&Snapshot> {
507 if snapshot.id == id {
508 return Some(snapshot);
509 }
510 for child in &snapshot.children {
511 if let Some(found) = find_node(child, id) {
512 return Some(found);
513 }
514 }
515 None
516}
517
518struct DiagnosticChild<'a> {
520 inner: Child<'a>,
521 meter: MeterOwner,
522}
523
524impl ChildPulse for DiagnosticChild<'_> {
525 fn complete_as(self: Box<Self>, outcome: Outcome) {
526 let Self { inner, meter } = *self;
527 how_far::Complete::complete_as(inner, outcome);
528 meter.close(outcome);
529 }
530 fn complete_inferred(self: Box<Self>, outcome: Outcome) {
531 let Self { inner, meter } = *self;
532 inner.complete_inferred(outcome);
533 meter.close(outcome);
534 }
535
536 fn finish(self: Box<Self>, outcome: Outcome) -> Result<(), PlanError> {
537 let Self { inner, meter } = *self;
538 let result = inner.finish(outcome);
539 meter.close(if result.is_ok() {
540 outcome
541 } else {
542 Outcome::Abandoned
543 });
544 result
545 }
546}
547
548struct DiagnosticView {
552 inner: how_far::SharedPulse,
553 meter: Arc<Meter>,
554}
555macro_rules! measured {
558 ($($wrapper:ty),*) => {$(
559 impl Stop for $wrapper {
560 #[track_caller]
561 fn check(&self) -> Result<(), StopReason> {
562 self.meter.check(&self.inner)
563 }
564 fn may_stop(&self) -> bool {
565 true
566 }
567 }
568 impl Report for $wrapper {
569 #[track_caller]
570 fn advance(&self, completed: u64) {
571 self.meter.advance(&self.inner, completed);
572 }
573 fn may_report(&self) -> bool {
574 true
575 }
576 }
577 impl Pulse for $wrapper {
578 #[track_caller]
579 fn record_issue(&self, error: PlanError) {
580 self.meter.issue(Problem::Plan(error));
581 }
582 fn split(
583 &self,
584 execution: Execution,
585 parts: &[PhaseSpec<'_>],
586 ) -> Result<Vec<Child<'_>>, PlanError> {
587 self.meter.split(&self.inner, execution, parts)
588 }
589 fn start(&self) -> Result<(), PlanError> {
590 self.inner.start()
591 }
592 fn set_total(&self, total: how_far::Total) -> Result<(), PlanError> {
593 self.inner.set_total(total)
594 }
595 fn share(&self) -> Result<how_far::SharedPulse, PlanError> {
596 Ok(how_far::SharedPulse::new(DiagnosticView {
597 inner: self.inner.share()?,
598 meter: Arc::clone(&self.meter),
599 }))
600 }
601 }
602 )*};
603}
604measured!(DiagnosticPulse, DiagnosticChild<'_>, DiagnosticView);
605
606impl Trace {
607 pub fn diagnose(&self, options: &Options) -> Vec<Finding> {
614 let mut findings = Vec::new();
615 for incident in &self.incidents {
616 findings.push(finding(match incident.problem { Problem::Plan(_) => Kind::ProtocolMisuse, _ => Kind::InvalidReport }, format!("phase {} at {}: {:?}", incident.node, incident.site, incident.problem), "Plan before reporting, select each phase once, and join workers before completing their owner. The work result was preserved."));
617 }
618 if self.dropped_incidents != 0 {
619 findings.push(finding(
620 Kind::IncompleteEvidence,
621 format!("{} reporting incidents omitted", self.dropped_incidents),
622 "Increase the incident budget or fix repeated protocol misuse.",
623 ));
624 }
625 if let Some(root) = &self.progress {
626 lifecycle_findings(root, false, &mut findings);
627 }
628 if let (Some(cancelled), Some(returned)) = (self.cancelled_at, self.returned_at)
630 && cancelled <= returned
631 && self.observed_at.is_none()
632 {
633 findings.push(finding(Kind::UnobservedCancellation, "host recorded cancellation and operation return, but no instrumented checkpoint observed the stop".into(), "Check at entry and finish the final paced batch. A request racing completion or uninstrumented checks can also explain this; timing alone does not prove a bug."));
634 }
635 if self.dropped_spans != 0 || self.active_spans != 0 {
636 findings.push(finding(
637 Kind::IncompleteEvidence,
638 format!(
639 "{} spans dropped; {} still active",
640 self.dropped_spans, self.active_spans
641 ),
642 "Increase profiler capacity and finish/join every task before diagnosing.",
643 ));
644 }
645 for span in &self.spans {
646 if span.stats.stopped_at.is_some() && span.outcome != Outcome::Cancelled {
647 findings.push(finding(Kind::StopOutcomeMismatch, format!("task {:?} observed a stop but ended {:?}", span.task, span.outcome), "Propagate StopReason and implement the borrowed error conversion. If recovery is intentional, record each attempt separately."));
648 }
649 if span.stats.overflowed
650 || span.stats.clock_regressions != 0
651 || span.stats.unattributed_calls != 0
652 {
653 findings.push(finding(Kind::IncompleteEvidence, format!(
654 "task {:?}: overflow={}, clock regressions={}, unattributed calls={}",
655 span.task, span.stats.overflowed, span.stats.clock_regressions,
656 span.stats.unattributed_calls
657 ), "Use a monotonic shared clock and sufficient span/site capacity before tuning this task."));
658 continue;
659 }
660 if span.kind != SpanKind::Work || span.outcome != Outcome::Succeeded {
661 continue;
662 }
663 if span.stats.max_check_gap > options.stop_gap_target
664 && span.elapsed() >= options.stop_gap_target
665 {
666 let mut ending = String::from("task exit");
669 for s in &span.stats.sites {
670 if s.max_gap_before_check == span.stats.max_check_gap {
671 ending = format!("{}:{}:{}", s.file, s.line, s.column);
672 }
673 }
674 let (covered, note) = cover(
675 self,
676 span,
677 span.stats.max_check_gap_start,
678 span.stats.max_check_gap,
679 options,
680 );
681 findings.push(finding(Kind::StopGap, format!(
682 "task {:?}: {:.2} ms without a check before {ending}{note}",
683 span.task,
684 ms(span.stats.max_check_gap),
685 ), if covered {
686 "This task's own checkpoints are sparse here, but the other task above checked at the target cadence throughout the interval. If it is the work inside this stage (for example an encoder holding its own Stop), cancellation is already covered and this is a measurement seam, not a latency problem. Otherwise add a cheap check() in the long-running section."
687 } else {
688 "Add a cheap check() in the long-running section; keep the stop-only cadence independent of reporting."
689 }));
690 }
691 if counts_units(self, span)
692 && let Some(gap) = &span.stats.max_report_gap
693 && gap.duration > options.report_gap_target
694 {
695 let own = gap.checks > 0 && gap.max_check_gap <= options.stop_gap_target;
696 let (covered, note) = if own {
697 (true, String::new())
698 } else {
699 cover(self, span, gap.start, gap.duration, options)
700 };
701 findings.push(finding(Kind::ReportGap, format!(
702 "task {:?}: {:.2} ms between {} and {}; stop checks inside it: {}, longest stop gap {:.2} ms{note}",
703 span.task, ms(gap.duration),
704 location(gap.from, "task entry"), location(gap.to, "task exit"),
705 gap.checks, ms(gap.max_check_gap)
706 ), if covered {
707 "Cancellation is checked at the target cadence inside this interval, so this is a progress-granularity seam rather than missing cancellation checks; more stop checks would not help. Finer progress needs a smaller reportable unit: sub-phases, an Estimated total, or a report from inside the long operation. Visible UI smoothness also depends on the consumer's polling cadence."
708 } else {
709 "No stop check at the target cadence was recorded inside this interval either. Consider reporting completed units and checking cancellation here; step(n) does both."
710 }));
711 }
712 if counts_units(self, span)
713 && span.stats.reports >= 8
714 && span.elapsed() >= options.minimum_stage_wall
715 && let Some(pace) = &span.stats.unit_pace
716 {
717 let q = &pace.quarters;
718 let fastest = q[0].min(q[1]).min(q[2]).min(q[3]);
719 let slowest = q[0].max(q[1]).max(q[2]).max(q[3]);
720 if !slowest.is_zero()
723 && slowest.as_secs_f64() >= fastest.as_secs_f64() * options.uneven_pace_ratio
724 {
725 findings.push(finding(Kind::UnevenPace, format!(
726 "task {:?}: the quarters of its {} units took {:.2}, {:.2}, {:.2} and {:.2} ms",
727 span.task, span.stats.units, ms(q[0]), ms(q[1]), ms(q[2]), ms(q[3])
728 ), "Progress, and display smoothing between reports, assume a phase's units cost about the same. If the change comes from the work itself (a different step), split the phase into stages where it changes, weighted by these times. If it comes from the input, report a unit that tracks the cost, such as bytes or pixels rather than items. The first quarter runs from the span's start: sequence entry for a sequential stage, otherwise its first check or report."));
729 }
730 }
731 #[cfg(feature = "experimental-stage-suggestions")]
732 {
733 if is_leaf_or_bare(self, span)
734 && span.elapsed() >= options.minimum_stage_wall
735 && let Some(found) = suggested_stages(span, options)
736 {
737 findings.push(found);
738 }
739 }
740 let seconds = span.elapsed().as_secs_f64();
741 if seconds > 0.0 {
742 for &(kind, threshold, advice) in &[
743 (
744 Kind::CheckFrequency,
745 options.check_calls_per_second,
746 "Measure the stop policy's actual cost before thinning checks; keep the required cancellation latency.",
747 ),
748 (
749 Kind::ReportFrequency,
750 options.report_calls_per_second,
751 "Consider batching reports per worker while keeping cancellation checks at their current cadence.",
752 ),
753 ] {
754 let sites = &span.stats.sites;
755 let calls = |index: usize| match kind {
756 Kind::CheckFrequency => sites[index].checks,
757 _ => sites[index].reports,
758 };
759 let order = sorted_indices(sites.len(), &|a, b| calls(b).cmp(&calls(a)));
761 for &index in &order[..order.len().min(3)] {
762 let site = &sites[index];
763 let calls = calls(index);
764 if calls < options.minimum_calls || (calls as f64 / seconds) <= threshold {
765 continue;
766 }
767 findings.push(finding(
768 kind,
769 format!(
770 "task {:?}: {} calls at {}:{}:{} in {:.2} ms ({:.0}/s)",
771 span.task,
772 calls,
773 site.file,
774 site.line,
775 site.column,
776 ms(span.elapsed()),
777 calls as f64 / seconds,
778 ),
779 advice,
780 ));
781 }
782 }
783 }
784 }
785 callback_findings(self, options, &mut findings);
786 if self.dropped_spans == 0
787 && self.active_spans == 0
788 && let Some(root) = &self.progress
789 {
790 stage_findings(self, root, options, &mut findings);
791 }
792 findings
793 }
794}
795
796fn cover(
802 trace: &Trace,
803 owner: &SpanRecord,
804 start: Duration,
805 len: Duration,
806 options: &Options,
807) -> (bool, String) {
808 let end = start.saturating_add(len);
809 let mut best: Option<&SpanRecord> = None;
810 for s in &trace.spans {
811 if s.id == owner.id
812 || s.kind != SpanKind::Work
813 || s.stats.checks == 0
814 || s.stats.overflowed
815 || s.stats.clock_regressions != 0
816 {
817 continue;
818 }
819 let overlap = s.end.min(end).saturating_sub(s.start.max(start));
820 if overlap.as_nanos() * 10 >= len.as_nanos() * 9
821 && best.is_none_or(|b| s.stats.max_check_gap < b.stats.max_check_gap)
822 {
823 best = Some(s);
824 }
825 }
826 match best {
827 Some(c) => (
829 c.stats.max_check_gap <= options.stop_gap_target,
830 format!(
831 "; task {:?} also ran across it with {} checks (longest gap {:.2} ms over its whole run)",
832 c.task,
833 c.stats.checks,
834 ms(c.stats.max_check_gap)
835 ),
836 ),
837 None => (false, String::new()),
838 }
839}
840
841struct CallbackGroup<'a> {
842 node: Option<NodeId>,
843 task: &'a str,
844 samples: Vec<(Duration, Duration)>,
845}
846
847fn busy_windows(trace: &Trace) -> Vec<(Duration, Duration)> {
849 let mut work: Vec<(Duration, Duration)> = Vec::new();
850 for span in &trace.spans {
851 if span.kind == SpanKind::Work {
852 work.push((span.start, span.end));
853 }
854 }
855 let mut merged: Vec<(Duration, Duration)> = Vec::new();
856 for &index in &sorted_indices(work.len(), &|a, b| work[a].0.cmp(&work[b].0)) {
857 let (start, end) = work[index];
858 match merged.last_mut() {
859 Some(last) if start <= last.1 => last.1 = last.1.max(end),
860 _ => merged.push((start, end)),
861 }
862 }
863 merged
864}
865
866fn within(windows: &[(Duration, Duration)], from: Duration, to: Duration) -> bool {
868 for &(start, end) in windows {
869 if start <= from && to <= end {
870 return true;
871 }
872 }
873 false
874}
875
876fn callback_findings(trace: &Trace, options: &Options, findings: &mut Vec<Finding>) {
877 let mut groups: Vec<CallbackGroup<'_>> = Vec::new();
878 'spans: for span in &trace.spans {
879 if span.kind != SpanKind::Callback || span.stats.clock_regressions != 0 {
880 continue;
881 }
882 let sample = (span.start, span.elapsed());
883 for group in &mut groups {
884 if group.node == span.node && group.task == span.task {
885 group.samples.push(sample);
886 continue 'spans;
887 }
888 }
889 groups.push(CallbackGroup {
890 node: span.node,
891 task: &span.task,
892 samples: vec![sample],
893 });
894 }
895 let busy = busy_windows(trace);
896 for CallbackGroup { task, samples, .. } in &groups {
897 let by_start = sorted_indices(samples.len(), &|a, b| samples[a].0.cmp(&samples[b].0));
898 let mut interval = Duration::ZERO;
899 for i in 1..by_start.len() {
900 let (from, to) = (samples[by_start[i - 1]].0, samples[by_start[i]].0);
901 if busy.is_empty() || within(&busy, from, to) {
903 interval = interval.max(to.saturating_sub(from));
904 }
905 }
906 if interval > options.callback_interval_target {
907 findings.push(finding(Kind::CallbackInterval, format!("callback {:?}: longest start-to-start interval {:.2} ms across {} retained calls (target {:.2} ms)",
908 task, ms(interval), samples.len(), ms(options.callback_interval_target)), "Poll or post notifications more regularly if the user interface needs this cadence; reporting alone does not dispatch callbacks."));
909 }
910 let by_duration = sorted_indices(samples.len(), &|a, b| samples[a].1.cmp(&samples[b].1));
911 let maximum = samples[by_duration[samples.len() - 1]].1; if maximum <= options.callback_budget {
913 continue;
914 }
915 let p95 = samples[by_duration[((samples.len() * 95).div_ceil(100)).saturating_sub(1)]].1;
916 findings.push(finding(Kind::CallbackDuration, format!("callback {:?}: max {:.2} ms, p95 {:.2} ms across {} retained calls (budget {:.2} ms)",
917 task, ms(maximum), ms(p95), samples.len(), ms(options.callback_budget)), "Keep subscriber work below the budget: defer rendering/I/O or post a coalesced notification to the owner thread. Time snapshot construction inside the callback if it happens there."));
918 }
919}
920
921fn phase_spec_code(child: &Snapshot, weight: u64) -> String {
922 let total = match child.initial_total {
923 how_far::Total::Exact(n) => format!("Total::Exact({n})"),
924 how_far::Total::Estimated(n) => format!("Total::Estimated({n})"),
925 _ => "Total::Unknown".into(),
926 };
927 let mut code = format!("PhaseSpec::new({:?}, {weight}, {total})", child.name);
928 if child.units != "items" {
929 code.push_str(&format!(".units({:?})", child.units));
930 }
931 match child.execution {
932 Execution::Unspecified => {}
933 Execution::Sequence => code.push_str(".execution(Execution::Sequence)"),
934 Execution::ForkJoin => code.push_str(".execution(Execution::ForkJoin)"),
935 Execution::WorkPool {
936 max_parallelism, ..
937 } => code.push_str(&format!(
938 ".execution(Execution::work_pool(NonZeroUsize::new({max_parallelism}).unwrap()))"
939 )),
940 _ => code.push_str(" /* preserve this stage's execution mode */"),
941 }
942 code
943}
944
945fn percent_weights(shares: &[f64]) -> Vec<u64> {
948 let mut weights = Vec::with_capacity(shares.len());
949 let mut assigned = 0_u64;
950 for share in shares {
951 let weight = (share * 100.0).floor().max(1.0) as u64;
952 weights.push(weight);
953 assigned += weight;
954 }
955 let remainder = |i: usize| shares[i] * 100.0 - (shares[i] * 100.0).floor();
956 let order = sorted_indices(shares.len(), &|a, b| remainder(b).total_cmp(&remainder(a)));
958 while assigned < 100 {
959 for &index in &order {
960 if assigned == 100 {
961 break;
962 }
963 weights[index] += 1;
964 assigned += 1;
965 }
966 }
967 while assigned > 100 {
968 let before = assigned;
969 for &index in order.iter().rev() {
970 if assigned == 100 {
971 break;
972 }
973 if weights[index] > 1 {
974 weights[index] -= 1;
975 assigned -= 1;
976 }
977 }
978 if assigned == before {
979 break; }
981 }
982 weights
983}
984
985fn stage_findings(
986 trace: &Trace,
987 parent: &Snapshot,
988 options: &Options,
989 findings: &mut Vec<Finding>,
990) {
991 if let Some(finding) = stage_weight_finding(trace, parent, options) {
992 findings.push(finding);
993 }
994 for child in &parent.children {
995 stage_findings(trace, child, options, findings);
996 }
997}
998
999fn stage_weight_finding(trace: &Trace, parent: &Snapshot, options: &Options) -> Option<Finding> {
1002 let children = &parent.children;
1003 if parent.execution != Execution::Sequence || children.len() < 2 || children.len() > 100 {
1004 return None;
1005 }
1006 let mut durations = Vec::with_capacity(children.len());
1007 let mut previous_end = Duration::ZERO;
1008 let mut total = Duration::ZERO;
1009 let mut weight_sum = 0_u64;
1010 for child in children {
1011 let (start, end) = stage_time(trace, child)?;
1012 if child.status != Status::Finished(Outcome::Succeeded) || previous_end > start {
1014 return None;
1015 }
1016 previous_end = end;
1017 let duration = end.saturating_sub(start);
1019 durations.push(duration);
1020 total = total.saturating_add(duration);
1021 weight_sum += child.weight;
1022 }
1023 if total < options.minimum_stage_wall || weight_sum == 0 {
1024 return None;
1025 }
1026 let mut shares = Vec::with_capacity(children.len());
1027 let mut declared = Vec::with_capacity(children.len());
1028 for i in 0..children.len() {
1029 shares.push(durations[i].as_secs_f64() / total.as_secs_f64());
1030 declared.push(children[i].weight as f64 / weight_sum as f64);
1031 }
1032 let mut frozen = 0_u128;
1039 let is_frozen = |frozen: u128, index: usize| frozen >> index & 1 == 1;
1040 let mut frozen_mass = 0.0;
1041 let negligible = sorted_indices(children.len(), &|a, b| declared[a].total_cmp(&declared[b]));
1042 for &index in &negligible {
1043 if shares[index] < options.negligible_stage_share && frozen_mass + declared[index] <= 0.5 {
1044 frozen |= 1 << index;
1045 frozen_mass += declared[index];
1046 }
1047 }
1048 let mut free_measured = 0.0;
1049 for (index, share) in shares.iter().enumerate() {
1050 if !is_frozen(frozen, index) {
1051 free_measured += share;
1052 }
1053 }
1054 let mut candidate = Vec::with_capacity(children.len());
1055 let mut biggest_difference = 0.0_f64;
1056 for i in 0..shares.len() {
1057 if is_frozen(frozen, i) {
1058 candidate.push(declared[i]);
1059 } else {
1060 let value = (1.0 - frozen_mass) * shares[i] / free_measured;
1061 biggest_difference = biggest_difference.max((declared[i] - value).abs());
1062 candidate.push(value);
1063 }
1064 }
1065 let advise = free_measured > 0.0 && biggest_difference >= options.weight_difference;
1066 if !advise {
1067 return None;
1068 }
1069 let weights = percent_weights(&candidate);
1070 let mut held = Vec::new();
1071 let mut planned = Vec::with_capacity(children.len());
1072 let mut code = String::from("[\n");
1073 for (i, child) in children.iter().enumerate() {
1074 if is_frozen(frozen, i) {
1075 held.push(child.name.as_str());
1076 }
1077 planned.push(child.weight);
1078 code.push_str(&format!(" {},\n", phase_spec_code(child, weights[i])));
1079 }
1080 code.push(']');
1081 let note = if held.is_empty() {
1082 String::new()
1083 } else {
1084 format!(
1085 "; {held:?} measured under {:.1}% of the run, too small to calibrate, so it keeps its declared share",
1086 options.negligible_stage_share * 100.0
1087 )
1088 };
1089 Some(Finding {
1090 kind: Kind::StageWeights,
1091 evidence: format!(
1092 "sequential stage {:?}: planned {planned:?}; measured wall-time candidate {weights:?} from one run{note}",
1093 parent.name
1094 ),
1095 advice: "Repeat across representative inputs before changing weights. Wall time includes waits and may shift with hardware, scheduling, or configuration.".into(),
1096 sample_code: Some(code),
1097 })
1098}
1099
1100#[cfg(feature = "experimental-stage-suggestions")]
1102fn is_leaf_or_bare(trace: &Trace, span: &SpanRecord) -> bool {
1103 match (span.node, &trace.progress) {
1104 (Some(node), Some(root)) => find_node(root, node).is_some_and(|p| p.children.is_empty()),
1105 _ => true,
1106 }
1107}
1108
1109#[cfg(feature = "experimental-stage-suggestions")]
1111#[derive(Clone, Copy)]
1112struct Stretch {
1113 start: Duration,
1114 end: Duration,
1115 time: Duration,
1116 busiest: Option<usize>,
1117 checks: u64,
1118 units: u64,
1119}
1120
1121#[cfg(feature = "experimental-stage-suggestions")]
1124fn suggested_stages(span: &SpanRecord, options: &Options) -> Option<Finding> {
1125 let stages = group_stages(span, options)?;
1126 Some(render_stages(span, &stages))
1127}
1128
1129#[cfg(feature = "experimental-stage-suggestions")]
1130struct GroupedStages {
1131 stretches: Vec<Stretch>,
1132 total: Duration,
1133 omitted: Duration,
1134 last_site: usize,
1135}
1136
1137#[cfg(feature = "experimental-stage-suggestions")]
1139fn group_stages(span: &SpanRecord, options: &Options) -> Option<GroupedStages> {
1140 let sites = &span.stats.sites;
1141 let window = |i: usize| match sites[i].active {
1148 Some((first, last)) => last.saturating_sub(first),
1149 None => Duration::MAX,
1150 };
1151 let order = sorted_indices(sites.len(), &|a, b| window(a).cmp(&window(b)));
1152 let mut found: Vec<Stretch> = Vec::new();
1153 let mut omitted = Duration::ZERO;
1155 let (mut last_site, mut last_call) = (None, span.start);
1156 for &i in &order {
1157 let site = &sites[i];
1158 let Some((first, last)) = site.active else {
1159 break;
1160 };
1161 if last_site.is_none() || last > last_call {
1162 (last_site, last_call) = (Some(i), last);
1163 }
1164 let overlaps = |s: &Stretch| first < s.end && s.start < last;
1165 let (mut count, mut covered) = (0, Duration::ZERO);
1166 for stretch in &found {
1167 if overlaps(stretch) {
1168 count += 1;
1169 covered += stretch.time;
1170 }
1171 }
1172 let mut next = Stretch {
1173 start: first,
1174 end: last,
1175 time: site.time,
1176 busiest: Some(i),
1177 checks: site.checks,
1178 units: site.units,
1179 };
1180 if count > 1 && site.time < covered {
1181 omitted += site.time;
1182 continue;
1183 }
1184 let mut k = 0;
1185 while k < found.len() {
1186 if overlaps(&found[k]) {
1187 let stretch = found.swap_remove(k);
1188 absorb(&mut next, stretch, sites);
1189 } else {
1190 k += 1;
1191 }
1192 }
1193 found.push(next);
1194 }
1195 let last_site = last_site?;
1196 let by_start = sorted_indices(found.len(), &|a, b| found[a].start.cmp(&found[b].start));
1199 let mut stretches = Vec::with_capacity(found.len() + 1);
1200 let mut total = Duration::ZERO;
1201 for &k in &by_start {
1202 total += found[k].time;
1203 stretches.push(found[k]);
1204 }
1205 let tail = span.end.saturating_sub(last_call);
1206 total += tail;
1207 stretches.push(Stretch {
1208 start: last_call,
1209 end: span.end,
1210 time: tail,
1211 busiest: None,
1212 checks: 0,
1213 units: 0,
1214 });
1215 let small = total.as_secs_f64() * options.negligible_stage_share;
1218 let mut kept = 0;
1219 for k in 0..stretches.len() {
1220 let next = stretches[k];
1221 if kept > 0
1222 && (next.time.as_secs_f64() < small
1223 || (kept == 1 && stretches[0].time.as_secs_f64() < small))
1224 {
1225 absorb(&mut stretches[kept - 1], next, sites);
1226 } else {
1227 stretches[kept] = next;
1228 kept += 1;
1229 }
1230 }
1231 if !matches!(kept, 2..=16) {
1232 return None;
1233 }
1234 stretches.truncate(kept);
1235 Some(GroupedStages {
1236 stretches,
1237 total,
1238 omitted,
1239 last_site,
1240 })
1241}
1242
1243#[cfg(feature = "experimental-stage-suggestions")]
1245fn render_stages(span: &SpanRecord, stages: &GroupedStages) -> Finding {
1246 use core::fmt::Write as _;
1247 let GroupedStages {
1248 stretches,
1249 total,
1250 omitted,
1251 last_site,
1252 } = stages;
1253 let sites = &span.stats.sites;
1254 let kept = stretches.len();
1255 let mut shares = Vec::with_capacity(kept);
1256 for stretch in stretches {
1257 shares.push(stretch.time.as_secs_f64() / total.as_secs_f64());
1258 }
1259 let weights = percent_weights(&shares);
1260 let evidence = if omitted.is_zero() {
1261 format!(
1262 "task {:?}: its checkpoints ran in {kept} stretches, one after another; the sample code lists them with this run's shares of time",
1263 span.task
1264 )
1265 } else {
1266 format!(
1267 "task {:?}: its checkpoints ran in {kept} stretches, one after another; {:.1} ms ({:.0}%) of the timed intervals ended at locations spanning several stretches (outer loops), and the weights spread that time over the stretches in proportion",
1268 span.task,
1269 ms(*omitted),
1270 100.0 * omitted.as_secs_f64() / (*total + *omitted).as_secs_f64()
1271 )
1272 };
1273 let mut code = String::from("Stages::new(pulse, &[\n");
1274 for k in 0..kept {
1275 let stretch = &stretches[k];
1276 let (start, end) = (
1277 ms(stretch.start.saturating_sub(span.start)),
1278 ms(stretch.end.saturating_sub(span.start)),
1279 );
1280 let tail = stretch.busiest.is_none();
1281 let site = &sites[stretch.busiest.unwrap_or(*last_site)];
1282 let name = format!(
1283 "{}{}:{}",
1284 if tail { "after " } else { "" },
1285 site.file,
1286 site.line
1287 );
1288 let _ = if tail {
1289 writeln!(
1290 code,
1291 " PhaseSpec::new({name:?}, {}, Total::Exact(1)), // {start:.1} to {end:.1} ms after the last timed check or report: add checks and units here",
1292 weights[k]
1293 )
1294 } else {
1295 let (count, note) = if stretch.units > 0 {
1296 (stretch.units, "units reported")
1297 } else {
1298 (stretch.checks, "checks: report a unit per check")
1299 };
1300 writeln!(
1301 code,
1302 " PhaseSpec::new({name:?}, {}, Total::Estimated({count})), // calls {start:.1} to {end:.1} ms; {:.1} ms of intervals end at them; {count} {note}",
1303 weights[k],
1304 ms(stretch.time)
1305 )
1306 };
1307 }
1308 code.push_str("])");
1309 Finding {
1310 kind: Kind::SuggestedStages,
1311 evidence,
1312 advice: "Declaring these as stages (Stages or Phases) lets the bar, and any smoothing of it, follow each part's own pace, and names each part in diagnostics. Give each stage a unit it can report, such as the loop iterations behind its checks. Each range is when a stretch's locations were called; the intervals its weight counts end at those calls, so its work starts before its range (setup before a first check counts toward the first stretch). Locations whose calls span several stretches (outer loops) are left out of the stretches and their time is spread over them. The weights are one run's shares of time: check them across inputs before relying on them.".into(),
1313 sample_code: Some(code),
1314 }
1315}
1316
1317#[cfg(feature = "experimental-stage-suggestions")]
1319fn absorb(stretch: &mut Stretch, next: Stretch, sites: &[crate::profile::SiteStats]) {
1320 stretch.start = stretch.start.min(next.start);
1321 stretch.end = stretch.end.max(next.end);
1322 stretch.time += next.time;
1323 stretch.checks = stretch.checks.saturating_add(next.checks);
1324 stretch.units = stretch.units.saturating_add(next.units);
1325 if let Some(next_site) = next.busiest
1326 && stretch
1327 .busiest
1328 .is_none_or(|current| sites[next_site].time > sites[current].time)
1329 {
1330 stretch.busiest = Some(next_site);
1331 }
1332}
1333
1334fn counts_units(trace: &Trace, span: &SpanRecord) -> bool {
1338 let (Some(node), Some(root)) = (span.node, &trace.progress) else {
1339 return span.stats.reports > 0;
1340 };
1341 find_node(root, node).is_some_and(|phase| phase.children.is_empty())
1342}
1343
1344fn stage_time(trace: &Trace, stage: &Snapshot) -> Option<(Duration, Duration)> {
1348 let mut time: Option<(Duration, Duration)> = None;
1349 for span in &trace.spans {
1350 let Some(node) = span.node else {
1351 continue;
1352 };
1353 if !matches!(span.kind, SpanKind::Work | SpanKind::Wait)
1354 || span.stats.clock_regressions != 0
1355 || span.outcome != Outcome::Succeeded
1356 || find_node(stage, node).is_none()
1357 {
1358 continue;
1359 }
1360 time = Some(match time {
1361 Some((start, end)) => (start.min(span.start), end.max(span.end)),
1362 None => (span.start, span.end),
1363 });
1364 }
1365 time
1366}
1367
1368impl how_far::Complete for DiagnosticPulse {
1369 fn complete_as(self, outcome: Outcome) {
1370 let Self { inner, meter } = self;
1371 how_far::Complete::complete_as(inner, outcome);
1372 meter.close(outcome);
1373 }
1374}
1375
1376fn lifecycle_findings(node: &Snapshot, parent_succeeded: bool, out: &mut Vec<Finding>) {
1377 let mut add = |kind, evidence: String, advice: &str| {
1378 out.push(finding(
1379 kind,
1380 format!("phase {:?} ({}): {}", node.name, node.id, evidence),
1381 advice,
1382 ))
1383 };
1384 if node.completion_inferred {
1385 add(
1386 Kind::InferredCompletion,
1387 format!("completion inferred as {:?}", node.status),
1388 "An unused optional phase and forgotten required work are indistinguishable from progress alone. Assert required outputs in library tests; skip explicitly when the decision is known.",
1389 );
1390 }
1391 if node.status == Status::Finished(Outcome::Abandoned) {
1392 add(
1393 Kind::AbandonedPhase,
1394 "owner or enclosing result boundary ended without this phase's result".into(),
1395 "Pass the final Result to complete or finish_phase. A bare ? bypasses that handoff; a panic or intentionally abandoned attempt can also cause this finding.",
1396 );
1397 }
1398 if parent_succeeded && node.status == Status::Finished(Outcome::Failed) {
1399 add(
1400 Kind::RecoveredFailure,
1401 "failed attempt inside a successful operation".into(),
1402 "This can be correct recovery. Keep the failed attempt visible and test that fallback fulfilled the operation's contract; the tracker cannot prove that.",
1403 );
1404 }
1405 if node.children.is_empty()
1406 && let how_far::Total::Exact(total) = node.total
1407 && (node.overrun()
1408 || (node.status == Status::Finished(Outcome::Succeeded) && node.completed != total))
1409 {
1410 add(
1411 Kind::CountMismatch,
1412 format!(
1413 "{} completed units for exact total {}",
1414 node.completed, total
1415 ),
1416 "Report completed work exactly once and flush local batches before completing the phase. If the total is not exact, declare Estimated or Unknown.",
1417 );
1418 }
1419 for child in &node.children {
1420 lifecycle_findings(
1421 child,
1422 node.status == Status::Finished(Outcome::Succeeded),
1423 out,
1424 );
1425 }
1426}