1#![forbid(unsafe_code)]
2
3use std::cmp::Ordering;
68use std::collections::BinaryHeap;
69use std::fmt::Write;
70
71const DEFAULT_AGING_FACTOR: f64 = 0.1;
73
74const MAX_QUEUE_SIZE: usize = 10_000;
76
77const DEFAULT_P_MIN_MS: f64 = 0.05;
79
80const DEFAULT_P_MAX_MS: f64 = 5_000.0;
82
83const DEFAULT_W_MIN: f64 = 1e-6;
85
86const DEFAULT_W_MAX: f64 = 100.0;
88
89const DEFAULT_WEIGHT_DEFAULT: f64 = 1.0;
91
92const DEFAULT_WEIGHT_UNKNOWN: f64 = 1.0;
94
95const DEFAULT_ESTIMATE_DEFAULT_MS: f64 = 10.0;
97
98const DEFAULT_ESTIMATE_UNKNOWN_MS: f64 = 1_000.0;
100
101const DEFAULT_WAIT_STARVE_MS: f64 = 500.0;
103
104const DEFAULT_STARVE_BOOST_RATIO: f64 = 1.5;
106
107#[derive(Debug, Clone)]
109pub struct SchedulerConfig {
110 pub aging_factor: f64,
114
115 pub p_min_ms: f64,
117
118 pub p_max_ms: f64,
120
121 pub estimate_default_ms: f64,
124
125 pub estimate_unknown_ms: f64,
128
129 pub w_min: f64,
131
132 pub w_max: f64,
134
135 pub weight_default: f64,
138
139 pub weight_unknown: f64,
142
143 pub wait_starve_ms: f64,
145
146 pub starve_boost_ratio: f64,
149
150 pub smith_enabled: bool,
152
153 pub force_fifo: bool,
156
157 pub max_queue_size: usize,
159
160 pub preemptive: bool,
162
163 pub time_quantum: f64,
166
167 pub enable_logging: bool,
169}
170
171impl Default for SchedulerConfig {
172 fn default() -> Self {
173 Self {
174 aging_factor: DEFAULT_AGING_FACTOR,
175 p_min_ms: DEFAULT_P_MIN_MS,
176 p_max_ms: DEFAULT_P_MAX_MS,
177 estimate_default_ms: DEFAULT_ESTIMATE_DEFAULT_MS,
178 estimate_unknown_ms: DEFAULT_ESTIMATE_UNKNOWN_MS,
179 w_min: DEFAULT_W_MIN,
180 w_max: DEFAULT_W_MAX,
181 weight_default: DEFAULT_WEIGHT_DEFAULT,
182 weight_unknown: DEFAULT_WEIGHT_UNKNOWN,
183 wait_starve_ms: DEFAULT_WAIT_STARVE_MS,
184 starve_boost_ratio: DEFAULT_STARVE_BOOST_RATIO,
185 smith_enabled: true,
186 force_fifo: false,
187 max_queue_size: MAX_QUEUE_SIZE,
188 preemptive: true,
189 time_quantum: 10.0,
190 enable_logging: false,
191 }
192 }
193}
194
195impl SchedulerConfig {
196 pub fn mode(&self) -> SchedulingMode {
198 if self.force_fifo {
199 SchedulingMode::Fifo
200 } else if self.smith_enabled {
201 SchedulingMode::Smith
202 } else {
203 SchedulingMode::Srpt
204 }
205 }
206}
207
208#[derive(Debug, Clone)]
210pub struct Job {
211 pub id: u64,
213
214 pub weight: f64,
216
217 pub remaining_time: f64,
219
220 pub total_time: f64,
222
223 pub arrival_time: f64,
225
226 pub arrival_seq: u64,
228
229 pub estimate_source: EstimateSource,
231
232 pub weight_source: WeightSource,
234
235 pub name: Option<String>,
237}
238
239impl Job {
240 pub fn new(id: u64, weight: f64, estimated_time: f64) -> Self {
242 let weight = if weight.is_nan() {
243 DEFAULT_W_MIN
244 } else if weight.is_infinite() {
245 if weight.is_sign_positive() {
246 DEFAULT_W_MAX
247 } else {
248 DEFAULT_W_MIN
249 }
250 } else {
251 weight.clamp(DEFAULT_W_MIN, DEFAULT_W_MAX)
252 };
253 let estimated_time = if estimated_time.is_nan() {
254 DEFAULT_P_MAX_MS
255 } else if estimated_time.is_infinite() {
256 if estimated_time.is_sign_positive() {
257 DEFAULT_P_MAX_MS
258 } else {
259 DEFAULT_P_MIN_MS
260 }
261 } else {
262 estimated_time.clamp(DEFAULT_P_MIN_MS, DEFAULT_P_MAX_MS)
263 };
264 Self {
265 id,
266 weight,
267 remaining_time: estimated_time,
268 total_time: estimated_time,
269 arrival_time: 0.0,
270 arrival_seq: 0,
271 estimate_source: EstimateSource::Explicit,
272 weight_source: WeightSource::Explicit,
273 name: None,
274 }
275 }
276
277 pub fn with_name(id: u64, weight: f64, estimated_time: f64, name: impl Into<String>) -> Self {
279 let mut job = Self::new(id, weight, estimated_time);
280 job.name = Some(name.into());
281 job
282 }
283
284 pub fn with_sources(
286 mut self,
287 weight_source: WeightSource,
288 estimate_source: EstimateSource,
289 ) -> Self {
290 self.weight_source = weight_source;
291 self.estimate_source = estimate_source;
292 self
293 }
294
295 pub fn progress(&self) -> f64 {
297 if self.total_time <= 0.0 {
298 1.0
299 } else {
300 1.0 - (self.remaining_time / self.total_time).clamp(0.0, 1.0)
301 }
302 }
303
304 pub fn is_complete(&self) -> bool {
306 self.remaining_time <= 0.0
307 }
308}
309
310#[derive(Debug, Clone)]
312struct PriorityJob {
313 priority: f64,
314 base_ratio: f64,
315 job: Job,
316 mode: SchedulingMode,
317}
318
319impl PartialEq for PriorityJob {
320 fn eq(&self, other: &Self) -> bool {
321 self.job.id == other.job.id
322 }
323}
324
325impl Eq for PriorityJob {}
326
327impl PartialOrd for PriorityJob {
328 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
329 Some(self.cmp(other))
330 }
331}
332
333impl Ord for PriorityJob {
334 fn cmp(&self, other: &Self) -> Ordering {
335 if self.mode == SchedulingMode::Fifo || other.mode == SchedulingMode::Fifo {
336 return other
337 .job
338 .arrival_seq
339 .cmp(&self.job.arrival_seq)
340 .then_with(|| other.job.id.cmp(&self.job.id));
341 }
342 self.priority
344 .total_cmp(&other.priority)
345 .then_with(|| self.base_ratio.total_cmp(&other.base_ratio))
347 .then_with(|| self.job.weight.total_cmp(&other.job.weight))
349 .then_with(|| other.job.remaining_time.total_cmp(&self.job.remaining_time))
351 .then_with(|| other.job.arrival_seq.cmp(&self.job.arrival_seq))
353 .then_with(|| other.job.id.cmp(&self.job.id))
355 }
356}
357
358#[derive(Debug, Clone)]
360pub struct SchedulingEvidence {
361 pub current_time: f64,
363
364 pub selected_job_id: Option<u64>,
366
367 pub queue_length: usize,
369
370 pub mean_wait_time: f64,
372
373 pub max_wait_time: f64,
375
376 pub reason: SelectionReason,
378
379 pub tie_break_reason: Option<TieBreakReason>,
381
382 pub jobs: Vec<JobEvidence>,
384}
385
386#[derive(Debug, Clone)]
388pub struct JobEvidence {
389 pub job_id: u64,
391 pub name: Option<String>,
393 pub estimate_ms: f64,
395 pub weight: f64,
397 pub ratio: f64,
399 pub aging_reward: f64,
401 pub starvation_floor: f64,
403 pub age_ms: f64,
405 pub effective_priority: f64,
407 pub objective_loss_proxy: f64,
409 pub estimate_source: EstimateSource,
411 pub weight_source: WeightSource,
413}
414
415#[derive(Debug, Clone, Copy, PartialEq, Eq)]
417pub enum SelectionReason {
418 QueueEmpty,
420 ShortestRemaining,
422 HighestWeightedPriority,
424 Fifo,
426 AgingBoost,
428 Continuation,
430}
431
432impl SelectionReason {
433 fn as_str(self) -> &'static str {
434 match self {
435 Self::QueueEmpty => "queue_empty",
436 Self::ShortestRemaining => "shortest_remaining",
437 Self::HighestWeightedPriority => "highest_weighted_priority",
438 Self::Fifo => "fifo",
439 Self::AgingBoost => "aging_boost",
440 Self::Continuation => "continuation",
441 }
442 }
443}
444
445#[derive(Debug, Clone, Copy, PartialEq, Eq)]
447pub enum EstimateSource {
448 Explicit,
450 Historical,
452 Default,
454 Unknown,
456}
457
458impl EstimateSource {
459 fn as_str(self) -> &'static str {
460 match self {
461 Self::Explicit => "explicit",
462 Self::Historical => "historical",
463 Self::Default => "default",
464 Self::Unknown => "unknown",
465 }
466 }
467}
468
469#[derive(Debug, Clone, Copy, PartialEq, Eq)]
471pub enum WeightSource {
472 Explicit,
474 Default,
476 Unknown,
478}
479
480impl WeightSource {
481 fn as_str(self) -> &'static str {
482 match self {
483 Self::Explicit => "explicit",
484 Self::Default => "default",
485 Self::Unknown => "unknown",
486 }
487 }
488}
489
490#[derive(Debug, Clone, Copy, PartialEq, Eq)]
492pub enum TieBreakReason {
493 EffectivePriority,
495 BaseRatio,
497 Weight,
499 RemainingTime,
501 ArrivalSeq,
503 JobId,
505 Continuation,
507}
508
509impl TieBreakReason {
510 fn as_str(self) -> &'static str {
511 match self {
512 Self::EffectivePriority => "effective_priority",
513 Self::BaseRatio => "base_ratio",
514 Self::Weight => "weight",
515 Self::RemainingTime => "remaining_time",
516 Self::ArrivalSeq => "arrival_seq",
517 Self::JobId => "job_id",
518 Self::Continuation => "continuation",
519 }
520 }
521}
522
523#[derive(Debug, Clone, Copy, PartialEq, Eq)]
525pub enum SchedulingMode {
526 Smith,
528 Srpt,
530 Fifo,
532}
533
534impl SchedulingEvidence {
535 #[must_use]
537 pub fn to_jsonl(&self, event: &str) -> String {
538 let mut out = String::with_capacity(256 + (self.jobs.len() * 64));
539 out.push_str("{\"event\":\"");
540 out.push_str(&escape_json(event));
541 out.push_str("\",\"current_time\":");
542 let _ = write!(out, "{:.6}", self.current_time);
543 out.push_str(",\"selected_job_id\":");
544 match self.selected_job_id {
545 Some(id) => {
546 let _ = write!(out, "{id}");
547 }
548 None => out.push_str("null"),
549 }
550 out.push_str(",\"queue_length\":");
551 let _ = write!(out, "{}", self.queue_length);
552 out.push_str(",\"mean_wait_time\":");
553 let _ = write!(out, "{:.6}", self.mean_wait_time);
554 out.push_str(",\"max_wait_time\":");
555 let _ = write!(out, "{:.6}", self.max_wait_time);
556 out.push_str(",\"reason\":\"");
557 out.push_str(self.reason.as_str());
558 out.push('"');
559 out.push_str(",\"tie_break_reason\":");
560 match self.tie_break_reason {
561 Some(reason) => {
562 out.push('"');
563 out.push_str(reason.as_str());
564 out.push('"');
565 }
566 None => out.push_str("null"),
567 }
568 out.push_str(",\"jobs\":[");
569 for (idx, job) in self.jobs.iter().enumerate() {
570 if idx > 0 {
571 out.push(',');
572 }
573 out.push_str(&job.to_json());
574 }
575 out.push_str("]}");
576 out
577 }
578}
579
580impl JobEvidence {
581 fn to_json(&self) -> String {
582 let mut out = String::with_capacity(128);
583 out.push_str("{\"job_id\":");
584 let _ = write!(out, "{}", self.job_id);
585 out.push_str(",\"name\":");
586 match &self.name {
587 Some(name) => {
588 out.push('"');
589 out.push_str(&escape_json(name));
590 out.push('"');
591 }
592 None => out.push_str("null"),
593 }
594 out.push_str(",\"estimate_ms\":");
595 let _ = write!(out, "{:.6}", self.estimate_ms);
596 out.push_str(",\"weight\":");
597 let _ = write!(out, "{:.6}", self.weight);
598 out.push_str(",\"ratio\":");
599 let _ = write!(out, "{:.6}", self.ratio);
600 out.push_str(",\"aging_reward\":");
601 let _ = write!(out, "{:.6}", self.aging_reward);
602 out.push_str(",\"starvation_floor\":");
603 let _ = write!(out, "{:.6}", self.starvation_floor);
604 out.push_str(",\"age_ms\":");
605 let _ = write!(out, "{:.6}", self.age_ms);
606 out.push_str(",\"effective_priority\":");
607 let _ = write!(out, "{:.6}", self.effective_priority);
608 out.push_str(",\"objective_loss_proxy\":");
609 let _ = write!(out, "{:.6}", self.objective_loss_proxy);
610 out.push_str(",\"estimate_source\":\"");
611 out.push_str(self.estimate_source.as_str());
612 out.push('"');
613 out.push_str(",\"weight_source\":\"");
614 out.push_str(self.weight_source.as_str());
615 out.push('"');
616 out.push('}');
617 out
618 }
619}
620
621fn escape_json(input: &str) -> String {
622 let mut out = String::with_capacity(input.len() + 8);
623 for ch in input.chars() {
624 match ch {
625 '"' => out.push_str("\\\""),
626 '\\' => out.push_str("\\\\"),
627 '\n' => out.push_str("\\n"),
628 '\r' => out.push_str("\\r"),
629 '\t' => out.push_str("\\t"),
630 '\u{08}' => out.push_str("\\b"),
631 '\u{0C}' => out.push_str("\\f"),
632 c if c < ' ' => {
633 let _ = write!(out, "\\u{:04x}", c as u32);
634 }
635 _ => out.push(ch),
636 }
637 }
638 out
639}
640
641#[derive(Debug, Clone, Default)]
643pub struct SchedulerStats {
644 pub total_submitted: u64,
646
647 pub total_completed: u64,
649
650 pub total_rejected: u64,
652
653 pub total_preemptions: u64,
655
656 pub total_processing_time: f64,
658
659 pub total_response_time: f64,
661
662 pub max_response_time: f64,
664
665 pub queue_length: usize,
667}
668
669impl SchedulerStats {
670 pub fn mean_response_time(&self) -> f64 {
672 if self.total_completed > 0 {
673 self.total_response_time / self.total_completed as f64
674 } else {
675 0.0
676 }
677 }
678
679 pub fn throughput(&self) -> f64 {
681 if self.total_processing_time > 0.0 {
682 self.total_completed as f64 / self.total_processing_time
683 } else {
684 0.0
685 }
686 }
687}
688
689#[derive(Debug)]
691pub struct QueueingScheduler {
692 config: SchedulerConfig,
693
694 queue: BinaryHeap<PriorityJob>,
696
697 current_job: Option<Job>,
699
700 current_time: f64,
702
703 next_job_id: u64,
705
706 next_arrival_seq: u64,
708
709 stats: SchedulerStats,
711}
712
713#[derive(Debug, Clone, Copy)]
714struct PriorityTerms {
715 aging_reward: f64,
716 starvation_floor: f64,
717 effective_priority: f64,
718}
719
720impl QueueingScheduler {
721 pub fn new(config: SchedulerConfig) -> Self {
723 Self {
724 config,
725 queue: BinaryHeap::new(),
726 current_job: None,
727 current_time: 0.0,
728 next_job_id: 1,
729 next_arrival_seq: 1,
730 stats: SchedulerStats::default(),
731 }
732 }
733
734 pub fn submit(&mut self, weight: f64, estimated_time: f64) -> Option<u64> {
738 self.submit_named(weight, estimated_time, None::<&str>)
739 }
740
741 pub fn submit_named(
743 &mut self,
744 weight: f64,
745 estimated_time: f64,
746 name: Option<impl Into<String>>,
747 ) -> Option<u64> {
748 self.submit_with_sources(
749 weight,
750 estimated_time,
751 WeightSource::Explicit,
752 EstimateSource::Explicit,
753 name,
754 )
755 }
756
757 pub fn submit_with_sources(
759 &mut self,
760 weight: f64,
761 estimated_time: f64,
762 weight_source: WeightSource,
763 estimate_source: EstimateSource,
764 name: Option<impl Into<String>>,
765 ) -> Option<u64> {
766 if self.queue.len() >= self.config.max_queue_size {
767 self.stats.total_rejected += 1;
768 return None;
769 }
770
771 let id = self.next_job_id;
772 self.next_job_id += 1;
773
774 let mut job = Job {
776 id,
777 weight,
778 remaining_time: estimated_time,
779 total_time: estimated_time,
780 arrival_time: 0.0,
781 arrival_seq: 0,
782 estimate_source,
783 weight_source,
784 name: None,
785 };
786 job.weight = self.normalize_weight_with_source(job.weight, job.weight_source);
787 job.remaining_time =
788 self.normalize_time_with_source(job.remaining_time, job.estimate_source);
789 job.total_time = job.remaining_time;
790 job.arrival_time = self.current_time;
791 job.arrival_seq = self.next_arrival_seq;
792 self.next_arrival_seq += 1;
793 if let Some(n) = name {
794 job.name = Some(n.into());
795 }
796
797 let priority_job = self.make_priority_job(job);
798 self.queue.push(priority_job);
799
800 self.stats.total_submitted += 1;
801 self.stats.queue_length = self.queue.len();
802
803 if self.config.preemptive {
805 self.maybe_preempt();
806 }
807
808 Some(id)
809 }
810
811 pub fn tick(&mut self, delta_time: f64) -> Vec<u64> {
815 let mut completed = Vec::new();
816 if !delta_time.is_finite() || delta_time <= 0.0 {
817 return completed;
818 }
819
820 let mut remaining_time = delta_time;
821 let mut now = self.current_time;
822 let mut processed_time = 0.0;
823
824 while remaining_time > 0.0 {
825 let Some(mut job) = (if let Some(j) = self.current_job.take() {
827 Some(j)
828 } else {
829 self.queue.pop().map(|pj| pj.job)
830 }) else {
831 now += remaining_time;
832 break; };
834
835 let process_time = remaining_time.min(job.remaining_time);
837 job.remaining_time -= process_time;
838 remaining_time -= process_time;
839 now += process_time;
840 processed_time += process_time;
841
842 if job.is_complete() {
843 let response_time = now - job.arrival_time;
845 self.stats.total_response_time += response_time;
846 self.stats.max_response_time = self.stats.max_response_time.max(response_time);
847 self.stats.total_completed += 1;
848 completed.push(job.id);
849 } else {
850 self.current_job = Some(job);
852 }
853 }
854
855 self.stats.total_processing_time += processed_time;
856 self.current_time = now;
857 self.refresh_priorities();
859
860 self.stats.queue_length = self.queue.len();
861 completed
862 }
863
864 pub fn peek_next(&self) -> Option<&Job> {
866 self.current_job
867 .as_ref()
868 .or_else(|| self.queue.peek().map(|pj| &pj.job))
869 }
870
871 pub fn evidence(&self) -> SchedulingEvidence {
873 let (mean_wait, max_wait) = self.compute_wait_stats();
874
875 let mut candidates: Vec<PriorityJob> = self
876 .queue
877 .iter()
878 .map(|pj| self.make_priority_job(pj.job.clone()))
879 .collect();
880
881 if let Some(ref current) = self.current_job {
882 candidates.push(self.make_priority_job(current.clone()));
883 }
884
885 candidates.sort_by(|a, b| b.cmp(a));
886
887 let selected_job_id = if let Some(ref current) = self.current_job {
888 Some(current.id)
889 } else {
890 candidates.first().map(|pj| pj.job.id)
891 };
892
893 let tie_break_reason = if self.current_job.is_some() {
894 Some(TieBreakReason::Continuation)
895 } else if candidates.len() > 1 {
896 Some(self.tie_break_reason(&candidates[0], &candidates[1]))
897 } else {
898 None
899 };
900
901 let reason = if self.queue.is_empty() && self.current_job.is_none() {
902 SelectionReason::QueueEmpty
903 } else if self.current_job.is_some() {
904 SelectionReason::Continuation
905 } else if self.config.mode() == SchedulingMode::Fifo {
906 SelectionReason::Fifo
907 } else if let Some(pj) = candidates.first() {
908 let wait_time = (self.current_time - pj.job.arrival_time).max(0.0);
909 let aging_contribution = self.config.aging_factor * wait_time;
910 let aging_boost = (self.config.wait_starve_ms > 0.0
911 && wait_time >= self.config.wait_starve_ms)
912 || aging_contribution > pj.base_ratio * 0.5;
913 if aging_boost {
914 SelectionReason::AgingBoost
915 } else if self.config.smith_enabled && pj.job.weight > 1.0 {
916 SelectionReason::HighestWeightedPriority
917 } else {
918 SelectionReason::ShortestRemaining
919 }
920 } else {
921 SelectionReason::QueueEmpty
922 };
923
924 let jobs = candidates
925 .iter()
926 .map(|pj| {
927 let age_ms = (self.current_time - pj.job.arrival_time).max(0.0);
928 let terms = self.compute_priority_terms(&pj.job);
929 JobEvidence {
930 job_id: pj.job.id,
931 name: pj.job.name.clone(),
932 estimate_ms: pj.job.remaining_time,
933 weight: pj.job.weight,
934 ratio: pj.base_ratio,
935 aging_reward: terms.aging_reward,
936 starvation_floor: terms.starvation_floor,
937 age_ms,
938 effective_priority: pj.priority,
939 objective_loss_proxy: 1.0 / pj.priority.max(self.config.w_min),
940 estimate_source: pj.job.estimate_source,
941 weight_source: pj.job.weight_source,
942 }
943 })
944 .collect();
945
946 SchedulingEvidence {
947 current_time: self.current_time,
948 selected_job_id,
949 queue_length: self.queue.len() + if self.current_job.is_some() { 1 } else { 0 },
950 mean_wait_time: mean_wait,
951 max_wait_time: max_wait,
952 reason,
953 tie_break_reason,
954 jobs,
955 }
956 }
957
958 pub fn stats(&self) -> SchedulerStats {
960 let mut stats = self.stats.clone();
961 stats.queue_length = self.queue.len() + if self.current_job.is_some() { 1 } else { 0 };
962 stats
963 }
964
965 pub fn cancel(&mut self, job_id: u64) -> bool {
967 if let Some(ref j) = self.current_job
969 && j.id == job_id
970 {
971 self.current_job = None;
972 self.stats.queue_length = self.queue.len();
973 return true;
974 }
975
976 let old_len = self.queue.len();
978 let jobs: Vec<_> = self
979 .queue
980 .drain()
981 .filter(|pj| pj.job.id != job_id)
982 .collect();
983 self.queue = jobs.into_iter().collect();
984
985 self.stats.queue_length = self.queue.len();
986 old_len != self.queue.len()
987 }
988
989 pub fn clear(&mut self) {
991 self.queue.clear();
992 self.current_job = None;
993 self.stats.queue_length = 0;
994 }
995
996 pub fn reset(&mut self) {
998 self.queue.clear();
999 self.current_job = None;
1000 self.current_time = 0.0;
1001 self.next_job_id = 1;
1002 self.next_arrival_seq = 1;
1003 self.stats = SchedulerStats::default();
1004 }
1005
1006 fn normalize_weight(&self, weight: f64) -> f64 {
1010 if weight.is_nan() {
1011 return self.config.w_min;
1012 }
1013 if weight.is_infinite() {
1014 return if weight.is_sign_positive() {
1015 self.config.w_max
1016 } else {
1017 self.config.w_min
1018 };
1019 }
1020 weight.clamp(self.config.w_min, self.config.w_max)
1021 }
1022
1023 fn normalize_time(&self, estimate_ms: f64) -> f64 {
1025 if estimate_ms.is_nan() {
1026 return self.config.p_max_ms;
1027 }
1028 if estimate_ms.is_infinite() {
1029 return if estimate_ms.is_sign_positive() {
1030 self.config.p_max_ms
1031 } else {
1032 self.config.p_min_ms
1033 };
1034 }
1035 estimate_ms.clamp(self.config.p_min_ms, self.config.p_max_ms)
1036 }
1037
1038 fn normalize_weight_with_source(&self, weight: f64, source: WeightSource) -> f64 {
1040 let resolved = match source {
1041 WeightSource::Explicit => weight,
1042 WeightSource::Default => self.config.weight_default,
1043 WeightSource::Unknown => self.config.weight_unknown,
1044 };
1045 self.normalize_weight(resolved)
1046 }
1047
1048 fn normalize_time_with_source(&self, estimate_ms: f64, source: EstimateSource) -> f64 {
1050 let resolved = match source {
1051 EstimateSource::Explicit | EstimateSource::Historical => estimate_ms,
1052 EstimateSource::Default => self.config.estimate_default_ms,
1053 EstimateSource::Unknown => self.config.estimate_unknown_ms,
1054 };
1055 self.normalize_time(resolved)
1056 }
1057
1058 fn compute_base_ratio(&self, job: &Job) -> f64 {
1060 if self.config.mode() == SchedulingMode::Fifo {
1061 return 0.0;
1062 }
1063 let remaining = job.remaining_time.max(self.config.p_min_ms);
1064 let weight = match self.config.mode() {
1065 SchedulingMode::Smith => job.weight,
1066 SchedulingMode::Srpt => 1.0,
1067 SchedulingMode::Fifo => 0.0,
1068 };
1069 weight / remaining
1070 }
1071
1072 fn compute_priority_terms(&self, job: &Job) -> PriorityTerms {
1080 if self.config.mode() == SchedulingMode::Fifo {
1081 return PriorityTerms {
1082 aging_reward: 0.0,
1083 starvation_floor: 0.0,
1084 effective_priority: 0.0,
1085 };
1086 }
1087
1088 let base_ratio = self.compute_base_ratio(job);
1089 let wait_time = (self.current_time - job.arrival_time).max(0.0);
1090 let aging_reward = self.config.aging_factor * wait_time;
1091 let starvation_floor =
1092 if self.config.wait_starve_ms > 0.0 && wait_time >= self.config.wait_starve_ms {
1093 base_ratio * self.config.starve_boost_ratio
1094 } else {
1095 0.0
1096 };
1097
1098 let effective_priority = (base_ratio + aging_reward).max(starvation_floor);
1099
1100 PriorityTerms {
1101 aging_reward,
1102 starvation_floor,
1103 effective_priority,
1104 }
1105 }
1106
1107 fn compute_priority(&self, job: &Job) -> f64 {
1109 self.compute_priority_terms(job).effective_priority
1110 }
1111
1112 fn make_priority_job(&self, job: Job) -> PriorityJob {
1114 let base_ratio = self.compute_base_ratio(&job);
1115 let priority = self.compute_priority(&job);
1116 PriorityJob {
1117 priority,
1118 base_ratio,
1119 job,
1120 mode: self.config.mode(),
1121 }
1122 }
1123
1124 fn tie_break_reason(&self, a: &PriorityJob, b: &PriorityJob) -> TieBreakReason {
1126 if self.config.mode() == SchedulingMode::Fifo {
1127 if a.job.arrival_seq != b.job.arrival_seq {
1128 return TieBreakReason::ArrivalSeq;
1129 }
1130 return TieBreakReason::JobId;
1131 }
1132 if a.priority.total_cmp(&b.priority) != Ordering::Equal {
1133 TieBreakReason::EffectivePriority
1134 } else if a.base_ratio.total_cmp(&b.base_ratio) != Ordering::Equal {
1135 TieBreakReason::BaseRatio
1136 } else if a.job.weight.total_cmp(&b.job.weight) != Ordering::Equal {
1137 TieBreakReason::Weight
1138 } else if a.job.remaining_time.total_cmp(&b.job.remaining_time) != Ordering::Equal {
1139 TieBreakReason::RemainingTime
1140 } else if a.job.arrival_seq != b.job.arrival_seq {
1141 TieBreakReason::ArrivalSeq
1142 } else {
1143 TieBreakReason::JobId
1144 }
1145 }
1146
1147 fn maybe_preempt(&mut self) {
1149 if self.config.mode() == SchedulingMode::Fifo {
1150 return;
1151 }
1152 if let Some(ref current) = self.current_job
1153 && let Some(pj) = self.queue.peek()
1154 {
1155 let current_pj = self.make_priority_job(current.clone());
1156 if pj.cmp(¤t_pj) == Ordering::Greater {
1157 let old = self
1159 .current_job
1160 .take()
1161 .expect("current_job guaranteed by if-let guard");
1162 let priority_job = self.make_priority_job(old);
1163 self.queue.push(priority_job);
1164 self.stats.total_preemptions += 1;
1165 }
1166 }
1167 }
1168
1169 fn refresh_priorities(&mut self) {
1171 let jobs: Vec<_> = self.queue.drain().map(|pj| pj.job).collect();
1172 for job in jobs {
1173 let priority_job = self.make_priority_job(job);
1174 self.queue.push(priority_job);
1175 }
1176 }
1177
1178 fn compute_wait_stats(&self) -> (f64, f64) {
1180 let mut total_wait = 0.0;
1181 let mut max_wait = 0.0f64;
1182 let mut count = 0;
1183
1184 for pj in self.queue.iter() {
1185 let wait = (self.current_time - pj.job.arrival_time).max(0.0);
1186 total_wait += wait;
1187 max_wait = max_wait.max(wait);
1188 count += 1;
1189 }
1190
1191 if let Some(ref j) = self.current_job {
1192 let wait = (self.current_time - j.arrival_time).max(0.0);
1193 total_wait += wait;
1194 max_wait = max_wait.max(wait);
1195 count += 1;
1196 }
1197
1198 let mean = if count > 0 {
1199 total_wait / count as f64
1200 } else {
1201 0.0
1202 };
1203 (mean, max_wait)
1204 }
1205}
1206
1207#[cfg(test)]
1212mod tests {
1213 use super::*;
1214 use std::collections::HashMap;
1215
1216 fn test_config() -> SchedulerConfig {
1217 SchedulerConfig {
1218 aging_factor: 0.001,
1219 p_min_ms: DEFAULT_P_MIN_MS,
1220 p_max_ms: DEFAULT_P_MAX_MS,
1221 estimate_default_ms: DEFAULT_ESTIMATE_DEFAULT_MS,
1222 estimate_unknown_ms: DEFAULT_ESTIMATE_UNKNOWN_MS,
1223 w_min: DEFAULT_W_MIN,
1224 w_max: DEFAULT_W_MAX,
1225 weight_default: DEFAULT_WEIGHT_DEFAULT,
1226 weight_unknown: DEFAULT_WEIGHT_UNKNOWN,
1227 wait_starve_ms: DEFAULT_WAIT_STARVE_MS,
1228 starve_boost_ratio: DEFAULT_STARVE_BOOST_RATIO,
1229 smith_enabled: true,
1230 force_fifo: false,
1231 max_queue_size: 100,
1232 preemptive: true,
1233 time_quantum: 10.0,
1234 enable_logging: false,
1235 }
1236 }
1237
1238 #[derive(Clone, Copy, Debug)]
1239 struct WorkloadJob {
1240 arrival: u64,
1241 weight: f64,
1242 duration: f64,
1243 }
1244
1245 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
1246 enum SimPolicy {
1247 Smith,
1248 Fifo,
1249 }
1250
1251 #[derive(Debug)]
1252 struct SimulationMetrics {
1253 mean: f64,
1254 p95: f64,
1255 p99: f64,
1256 max: f64,
1257 job_count: usize,
1258 completion_order: Vec<u64>,
1259 }
1260
1261 fn mixed_workload() -> Vec<WorkloadJob> {
1262 let mut jobs = Vec::new();
1263 jobs.push(WorkloadJob {
1264 arrival: 0,
1265 weight: 1.0,
1266 duration: 100.0,
1267 });
1268 for t in 1..=200u64 {
1269 jobs.push(WorkloadJob {
1270 arrival: t,
1271 weight: 1.0,
1272 duration: 1.0,
1273 });
1274 }
1275 jobs
1276 }
1277
1278 fn percentile(sorted: &[f64], p: f64) -> f64 {
1279 if sorted.is_empty() {
1280 return 0.0;
1281 }
1282 let idx = ((sorted.len() as f64 - 1.0) * p).ceil() as usize;
1283 sorted[idx.min(sorted.len() - 1)]
1284 }
1285
1286 fn summary_json(policy: SimPolicy, metrics: &SimulationMetrics) -> String {
1287 let policy = match policy {
1288 SimPolicy::Smith => "Smith",
1289 SimPolicy::Fifo => "Fifo",
1290 };
1291 let head: Vec<String> = metrics
1292 .completion_order
1293 .iter()
1294 .take(8)
1295 .map(|id| id.to_string())
1296 .collect();
1297 let tail: Vec<String> = metrics
1298 .completion_order
1299 .iter()
1300 .rev()
1301 .take(3)
1302 .collect::<Vec<_>>()
1303 .into_iter()
1304 .rev()
1305 .map(|id| id.to_string())
1306 .collect();
1307 format!(
1308 "{{\"policy\":\"{policy}\",\"jobs\":{jobs},\"mean\":{mean:.3},\"p95\":{p95:.3},\"p99\":{p99:.3},\"max\":{max:.3},\"order_head\":[{head}],\"order_tail\":[{tail}]}}",
1309 policy = policy,
1310 jobs = metrics.job_count,
1311 mean = metrics.mean,
1312 p95 = metrics.p95,
1313 p99 = metrics.p99,
1314 max = metrics.max,
1315 head = head.join(","),
1316 tail = tail.join(",")
1317 )
1318 }
1319
1320 fn workload_summary_json(workload: &[WorkloadJob]) -> String {
1321 if workload.is_empty() {
1322 return "{\"workload\":\"empty\"}".to_string();
1323 }
1324 let mut min_arrival = u64::MAX;
1325 let mut max_arrival = 0u64;
1326 let mut min_duration = f64::INFINITY;
1327 let mut max_duration: f64 = 0.0;
1328 let mut total_work: f64 = 0.0;
1329 let mut long_jobs = 0usize;
1330 let long_threshold = 10.0;
1331
1332 for job in workload {
1333 min_arrival = min_arrival.min(job.arrival);
1334 max_arrival = max_arrival.max(job.arrival);
1335 min_duration = min_duration.min(job.duration);
1336 max_duration = max_duration.max(job.duration);
1337 total_work += job.duration;
1338 if job.duration >= long_threshold {
1339 long_jobs += 1;
1340 }
1341 }
1342
1343 format!(
1344 "{{\"workload\":\"mixed\",\"jobs\":{jobs},\"arrival_min\":{arrival_min},\"arrival_max\":{arrival_max},\"duration_min\":{duration_min:.3},\"duration_max\":{duration_max:.3},\"total_work\":{total_work:.3},\"long_jobs\":{long_jobs},\"long_threshold\":{long_threshold:.1}}}",
1345 jobs = workload.len(),
1346 arrival_min = min_arrival,
1347 arrival_max = max_arrival,
1348 duration_min = min_duration,
1349 duration_max = max_duration,
1350 total_work = total_work,
1351 long_jobs = long_jobs,
1352 long_threshold = long_threshold
1353 )
1354 }
1355
1356 fn simulate_policy(policy: SimPolicy, workload: &[WorkloadJob]) -> SimulationMetrics {
1357 let mut config = test_config();
1358 config.aging_factor = 0.0;
1359 config.wait_starve_ms = 0.0;
1360 config.starve_boost_ratio = 1.0;
1361 config.smith_enabled = policy == SimPolicy::Smith;
1362 config.force_fifo = policy == SimPolicy::Fifo;
1363 config.preemptive = true;
1364
1365 let mut scheduler = QueueingScheduler::new(config);
1366 let mut arrivals = workload.to_vec();
1367 arrivals.sort_by_key(|job| job.arrival);
1368
1369 let mut arrival_times: HashMap<u64, f64> = HashMap::new();
1370 let mut response_times = Vec::with_capacity(arrivals.len());
1371 let mut completion_order = Vec::with_capacity(arrivals.len());
1372
1373 let mut idx = 0usize;
1374 let mut safety = 0usize;
1375
1376 while (idx < arrivals.len() || scheduler.peek_next().is_some()) && safety < 10_000 {
1377 let now = scheduler.current_time;
1378
1379 while idx < arrivals.len() && (arrivals[idx].arrival as f64) <= now + f64::EPSILON {
1380 let job = arrivals[idx];
1381 let id = scheduler
1382 .submit(job.weight, job.duration)
1383 .expect("queue capacity should not be exceeded");
1384 arrival_times.insert(id, scheduler.current_time);
1385 idx += 1;
1386 }
1387
1388 if scheduler.peek_next().is_none() {
1389 if idx < arrivals.len() {
1390 let next_time = arrivals[idx].arrival as f64;
1391 let delta = (next_time - scheduler.current_time).max(0.0);
1392 let completed = scheduler.tick(delta);
1393 for id in completed {
1394 let arrival = arrival_times.get(&id).copied().unwrap_or(0.0);
1395 response_times.push(scheduler.current_time - arrival);
1396 completion_order.push(id);
1397 }
1398 }
1399 safety += 1;
1400 continue;
1401 }
1402
1403 let completed = scheduler.tick(1.0);
1404 for id in completed {
1405 let arrival = arrival_times.get(&id).copied().unwrap_or(0.0);
1406 response_times.push(scheduler.current_time - arrival);
1407 completion_order.push(id);
1408 }
1409 safety += 1;
1410 }
1411
1412 assert_eq!(
1413 response_times.len(),
1414 arrivals.len(),
1415 "simulation did not complete all jobs"
1416 );
1417
1418 let mut sorted = response_times.clone();
1419 sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());
1420
1421 let mean = response_times.iter().sum::<f64>() / response_times.len() as f64;
1422 let p95 = percentile(&sorted, 0.95);
1423 let p99 = percentile(&sorted, 0.99);
1424 let max = *sorted.last().unwrap_or(&0.0);
1425
1426 SimulationMetrics {
1427 mean,
1428 p95,
1429 p99,
1430 max,
1431 job_count: response_times.len(),
1432 completion_order,
1433 }
1434 }
1435
1436 #[test]
1441 fn new_creates_empty_scheduler() {
1442 let scheduler = QueueingScheduler::new(test_config());
1443 assert_eq!(scheduler.stats().queue_length, 0);
1444 assert!(scheduler.peek_next().is_none());
1445 }
1446
1447 #[test]
1448 fn default_config_valid() {
1449 let config = SchedulerConfig::default();
1450 let scheduler = QueueingScheduler::new(config);
1451 assert_eq!(scheduler.stats().queue_length, 0);
1452 }
1453
1454 #[test]
1459 fn submit_returns_job_id() {
1460 let mut scheduler = QueueingScheduler::new(test_config());
1461 let id = scheduler.submit(1.0, 10.0);
1462 assert_eq!(id, Some(1));
1463 }
1464
1465 #[test]
1466 fn submit_increments_job_id() {
1467 let mut scheduler = QueueingScheduler::new(test_config());
1468 let id1 = scheduler.submit(1.0, 10.0);
1469 let id2 = scheduler.submit(1.0, 10.0);
1470 assert_eq!(id1, Some(1));
1471 assert_eq!(id2, Some(2));
1472 }
1473
1474 #[test]
1475 fn submit_rejects_when_queue_full() {
1476 let mut config = test_config();
1477 config.max_queue_size = 2;
1478 let mut scheduler = QueueingScheduler::new(config);
1479
1480 assert!(scheduler.submit(1.0, 10.0).is_some());
1481 assert!(scheduler.submit(1.0, 10.0).is_some());
1482 assert!(scheduler.submit(1.0, 10.0).is_none()); assert_eq!(scheduler.stats().total_rejected, 1);
1484 }
1485
1486 #[test]
1487 fn submit_named_job() {
1488 let mut scheduler = QueueingScheduler::new(test_config());
1489 let id = scheduler.submit_named(1.0, 10.0, Some("test-job"));
1490 assert!(id.is_some());
1491 }
1492
1493 #[test]
1498 fn srpt_prefers_shorter_jobs() {
1499 let mut scheduler = QueueingScheduler::new(test_config());
1500
1501 scheduler.submit(1.0, 100.0); scheduler.submit(1.0, 10.0); let next = scheduler.peek_next().unwrap();
1505 assert_eq!(next.remaining_time, 10.0); }
1507
1508 #[test]
1509 fn smith_rule_prefers_high_weight() {
1510 let mut scheduler = QueueingScheduler::new(test_config());
1511
1512 scheduler.submit(1.0, 10.0); scheduler.submit(10.0, 10.0); let next = scheduler.peek_next().unwrap();
1516 assert_eq!(next.weight, 10.0); }
1518
1519 #[test]
1520 fn smith_rule_balances_weight_and_time() {
1521 let mut scheduler = QueueingScheduler::new(test_config());
1522
1523 scheduler.submit(2.0, 20.0); scheduler.submit(1.0, 5.0); let next = scheduler.peek_next().unwrap();
1527 assert_eq!(next.remaining_time, 5.0); }
1529
1530 #[test]
1535 fn aging_increases_priority_over_time() {
1536 let mut scheduler = QueueingScheduler::new(test_config());
1537
1538 scheduler.submit(1.0, 100.0); scheduler.tick(0.0); let before_aging = scheduler.compute_priority(scheduler.peek_next().unwrap());
1542
1543 scheduler.current_time = 100.0; scheduler.refresh_priorities();
1545
1546 let after_aging = scheduler.compute_priority(scheduler.peek_next().unwrap());
1547 assert!(
1548 after_aging > before_aging,
1549 "Priority should increase with wait time"
1550 );
1551 }
1552
1553 #[test]
1554 fn aging_prevents_starvation() {
1555 let mut config = test_config();
1556 config.aging_factor = 1.0; let mut scheduler = QueueingScheduler::new(config);
1558
1559 scheduler.submit(1.0, 1000.0); scheduler.submit(1.0, 1.0); assert_eq!(scheduler.peek_next().unwrap().remaining_time, 1.0);
1564
1565 let completed = scheduler.tick(1.0);
1567 assert_eq!(completed.len(), 1);
1568
1569 assert!(scheduler.peek_next().is_some());
1570 }
1571
1572 #[test]
1577 fn preemption_when_higher_priority_arrives() {
1578 let mut scheduler = QueueingScheduler::new(test_config());
1579
1580 scheduler.submit(1.0, 100.0); scheduler.tick(10.0); let before = scheduler.peek_next().unwrap().remaining_time;
1584 assert_eq!(before, 90.0);
1585
1586 scheduler.submit(1.0, 5.0); let next = scheduler.peek_next().unwrap();
1590 assert_eq!(next.remaining_time, 5.0);
1591
1592 assert_eq!(scheduler.stats().total_preemptions, 1);
1594 }
1595
1596 #[test]
1597 fn no_preemption_when_disabled() {
1598 let mut config = test_config();
1599 config.preemptive = false;
1600 let mut scheduler = QueueingScheduler::new(config);
1601
1602 scheduler.submit(1.0, 100.0);
1603 scheduler.tick(10.0);
1604
1605 scheduler.submit(1.0, 5.0); let next = scheduler.peek_next().unwrap();
1609 assert_eq!(next.remaining_time, 90.0);
1610 }
1611
1612 #[test]
1617 fn tick_processes_jobs() {
1618 let mut scheduler = QueueingScheduler::new(test_config());
1619
1620 scheduler.submit(1.0, 10.0);
1621 let completed = scheduler.tick(5.0);
1622
1623 assert!(completed.is_empty()); assert_eq!(scheduler.peek_next().unwrap().remaining_time, 5.0);
1625 }
1626
1627 #[test]
1628 fn tick_completes_jobs() {
1629 let mut scheduler = QueueingScheduler::new(test_config());
1630
1631 scheduler.submit(1.0, 10.0);
1632 let completed = scheduler.tick(10.0);
1633
1634 assert_eq!(completed.len(), 1);
1635 assert_eq!(completed[0], 1);
1636 assert!(scheduler.peek_next().is_none());
1637 }
1638
1639 #[test]
1640 fn tick_completes_multiple_jobs() {
1641 let mut scheduler = QueueingScheduler::new(test_config());
1642
1643 scheduler.submit(1.0, 5.0);
1644 scheduler.submit(1.0, 5.0);
1645 let completed = scheduler.tick(10.0);
1646
1647 assert_eq!(completed.len(), 2);
1648 }
1649
1650 #[test]
1651 fn tick_handles_zero_delta() {
1652 let mut scheduler = QueueingScheduler::new(test_config());
1653 scheduler.submit(1.0, 10.0);
1654 let completed = scheduler.tick(0.0);
1655 assert!(completed.is_empty());
1656 }
1657
1658 #[test]
1663 fn stats_track_submissions() {
1664 let mut scheduler = QueueingScheduler::new(test_config());
1665
1666 scheduler.submit(1.0, 10.0);
1667 scheduler.submit(1.0, 10.0);
1668
1669 let stats = scheduler.stats();
1670 assert_eq!(stats.total_submitted, 2);
1671 assert_eq!(stats.queue_length, 2);
1672 }
1673
1674 #[test]
1675 fn stats_track_completions() {
1676 let mut scheduler = QueueingScheduler::new(test_config());
1677
1678 scheduler.submit(1.0, 10.0);
1679 scheduler.tick(10.0);
1680
1681 let stats = scheduler.stats();
1682 assert_eq!(stats.total_completed, 1);
1683 }
1684
1685 #[test]
1686 fn stats_compute_mean_response_time() {
1687 let mut scheduler = QueueingScheduler::new(test_config());
1688
1689 scheduler.submit(1.0, 10.0);
1690 scheduler.submit(1.0, 10.0);
1691 scheduler.tick(20.0);
1692
1693 let stats = scheduler.stats();
1694 assert_eq!(stats.total_completed, 2);
1697 assert!(stats.mean_response_time() > 0.0);
1698 }
1699
1700 #[test]
1701 fn stats_compute_throughput() {
1702 let mut scheduler = QueueingScheduler::new(test_config());
1703
1704 scheduler.submit(1.0, 10.0);
1705 scheduler.tick(10.0);
1706
1707 let stats = scheduler.stats();
1708 assert!((stats.throughput() - 0.1).abs() < 0.01);
1710 }
1711
1712 #[test]
1717 fn evidence_reports_queue_empty() {
1718 let scheduler = QueueingScheduler::new(test_config());
1719 let evidence = scheduler.evidence();
1720 assert_eq!(evidence.reason, SelectionReason::QueueEmpty);
1721 assert!(evidence.selected_job_id.is_none());
1722 assert!(evidence.tie_break_reason.is_none());
1723 assert!(evidence.jobs.is_empty());
1724 }
1725
1726 #[test]
1727 fn evidence_reports_selected_job() {
1728 let mut scheduler = QueueingScheduler::new(test_config());
1729 scheduler.submit(1.0, 10.0);
1730 let evidence = scheduler.evidence();
1731 assert_eq!(evidence.selected_job_id, Some(1));
1732 assert_eq!(evidence.jobs.len(), 1);
1733 }
1734
1735 #[test]
1736 fn evidence_reports_wait_stats() {
1737 let mut scheduler = QueueingScheduler::new(test_config());
1738 scheduler.submit(1.0, 100.0);
1739 scheduler.submit(1.0, 100.0);
1740 scheduler.current_time = 50.0;
1741 scheduler.refresh_priorities();
1742
1743 let evidence = scheduler.evidence();
1744 assert!(evidence.mean_wait_time > 0.0);
1745 assert!(evidence.max_wait_time > 0.0);
1746 }
1747
1748 #[test]
1749 fn evidence_reports_priority_objective_terms() {
1750 let mut config = test_config();
1751 config.aging_factor = 0.5;
1752 config.wait_starve_ms = 10.0;
1753 config.starve_boost_ratio = 2.0;
1754 let mut scheduler = QueueingScheduler::new(config);
1755
1756 scheduler.submit(1.0, 20.0);
1757 scheduler.current_time = 20.0;
1758 scheduler.refresh_priorities();
1759
1760 let evidence = scheduler.evidence();
1761 let job = evidence.jobs.first().expect("job evidence");
1762 assert!(job.aging_reward > 0.0);
1763 assert!(job.starvation_floor > 0.0);
1764 assert!(job.effective_priority >= job.ratio + job.aging_reward);
1765 assert!(
1766 (job.objective_loss_proxy - (1.0 / job.effective_priority.max(DEFAULT_W_MIN))).abs()
1767 < 1e-9
1768 );
1769 }
1770
1771 #[test]
1776 fn force_fifo_overrides_priority() {
1777 let mut config = test_config();
1778 config.force_fifo = true;
1779 let mut scheduler = QueueingScheduler::new(config);
1780
1781 let first = scheduler.submit(1.0, 100.0).unwrap();
1782 let second = scheduler.submit(10.0, 1.0).unwrap();
1783
1784 let next = scheduler.peek_next().unwrap();
1785 assert_eq!(next.id, first);
1786 assert_ne!(next.id, second);
1787 assert_eq!(scheduler.evidence().reason, SelectionReason::Fifo);
1788 }
1789
1790 #[test]
1791 fn default_sources_use_config_values() {
1792 let mut config = test_config();
1793 config.weight_default = 7.0;
1794 config.estimate_default_ms = 12.0;
1795 let mut scheduler = QueueingScheduler::new(config);
1796
1797 scheduler.submit_with_sources(
1798 999.0,
1799 999.0,
1800 WeightSource::Default,
1801 EstimateSource::Default,
1802 None::<&str>,
1803 );
1804
1805 let next = scheduler.peek_next().unwrap();
1806 assert!((next.weight - 7.0).abs() < f64::EPSILON);
1807 assert!((next.remaining_time - 12.0).abs() < f64::EPSILON);
1808 }
1809
1810 #[test]
1811 fn unknown_sources_use_config_values() {
1812 let mut config = test_config();
1813 config.weight_unknown = 2.5;
1814 config.estimate_unknown_ms = 250.0;
1815 let mut scheduler = QueueingScheduler::new(config);
1816
1817 scheduler.submit_with_sources(
1818 0.0,
1819 0.0,
1820 WeightSource::Unknown,
1821 EstimateSource::Unknown,
1822 None::<&str>,
1823 );
1824
1825 let next = scheduler.peek_next().unwrap();
1826 assert!((next.weight - 2.5).abs() < f64::EPSILON);
1827 assert!((next.remaining_time - 250.0).abs() < f64::EPSILON);
1828 }
1829
1830 #[test]
1835 fn tie_break_prefers_base_ratio_when_effective_equal() {
1836 let mut config = test_config();
1837 config.aging_factor = 0.1;
1838 let mut scheduler = QueueingScheduler::new(config);
1839
1840 let id_a = scheduler.submit(1.0, 2.0).unwrap(); scheduler.current_time = 5.0;
1843 scheduler.refresh_priorities();
1844
1845 let id_b = scheduler.submit(1.0, 1.0).unwrap(); scheduler.refresh_priorities();
1848
1849 let next = scheduler.peek_next().unwrap();
1850 assert_eq!(next.id, id_b);
1851
1852 let evidence = scheduler.evidence();
1853 assert_eq!(evidence.selected_job_id, Some(id_b));
1854 assert_eq!(evidence.tie_break_reason, Some(TieBreakReason::BaseRatio));
1855 assert_ne!(id_a, id_b);
1856 }
1857
1858 #[test]
1859 fn tie_break_prefers_weight_over_arrival() {
1860 let mut scheduler = QueueingScheduler::new(test_config());
1861
1862 let high_weight = scheduler.submit(2.0, 2.0).unwrap(); let _low_weight = scheduler.submit(1.0, 1.0).unwrap(); let evidence = scheduler.evidence();
1866 assert_eq!(evidence.selected_job_id, Some(high_weight));
1867 assert_eq!(evidence.tie_break_reason, Some(TieBreakReason::Weight));
1868 }
1869
1870 #[test]
1871 fn tie_break_prefers_arrival_seq_when_all_equal() {
1872 let mut config = test_config();
1873 config.aging_factor = 0.0;
1874 let mut scheduler = QueueingScheduler::new(config);
1875
1876 let first = scheduler.submit(1.0, 10.0).unwrap();
1877 let second = scheduler.submit(1.0, 10.0).unwrap();
1878
1879 let evidence = scheduler.evidence();
1880 assert_eq!(evidence.selected_job_id, Some(first));
1881 assert_eq!(evidence.tie_break_reason, Some(TieBreakReason::ArrivalSeq));
1882 assert_ne!(first, second);
1883 }
1884
1885 #[test]
1890 fn srpt_mode_ignores_weights() {
1891 let mut config = test_config();
1892 config.smith_enabled = false;
1893 let mut scheduler = QueueingScheduler::new(config);
1894
1895 scheduler.submit(10.0, 100.0); scheduler.submit(1.0, 10.0); let next = scheduler.peek_next().unwrap();
1899 assert_eq!(next.remaining_time, 10.0);
1900 assert_eq!(
1901 scheduler.evidence().reason,
1902 SelectionReason::ShortestRemaining
1903 );
1904 }
1905
1906 #[test]
1907 fn fifo_mode_disables_preemption() {
1908 let mut config = test_config();
1909 config.force_fifo = true;
1910 config.preemptive = true;
1911 let mut scheduler = QueueingScheduler::new(config);
1912
1913 let first = scheduler.submit(1.0, 100.0).unwrap();
1914 scheduler.tick(10.0);
1915
1916 let _later = scheduler.submit(10.0, 1.0).unwrap();
1917 let next = scheduler.peek_next().unwrap();
1918 assert_eq!(next.id, first);
1919 }
1920
1921 #[test]
1922 fn explicit_zero_weight_clamps_to_min() {
1923 let mut config = test_config();
1924 config.w_min = 0.5;
1925 let mut scheduler = QueueingScheduler::new(config);
1926
1927 scheduler.submit_with_sources(
1928 0.0,
1929 1.0,
1930 WeightSource::Explicit,
1931 EstimateSource::Explicit,
1932 None::<&str>,
1933 );
1934
1935 let next = scheduler.peek_next().unwrap();
1936 assert!((next.weight - 0.5).abs() < f64::EPSILON);
1937 }
1938
1939 #[test]
1940 fn explicit_zero_estimate_clamps_to_min() {
1941 let mut config = test_config();
1942 config.p_min_ms = 2.0;
1943 let mut scheduler = QueueingScheduler::new(config);
1944
1945 scheduler.submit_with_sources(
1946 1.0,
1947 0.0,
1948 WeightSource::Explicit,
1949 EstimateSource::Explicit,
1950 None::<&str>,
1951 );
1952
1953 let next = scheduler.peek_next().unwrap();
1954 assert!((next.remaining_time - 2.0).abs() < f64::EPSILON);
1955 }
1956
1957 #[test]
1958 fn explicit_weight_honors_config_w_max_above_defaults() {
1959 let mut config = test_config();
1960 config.w_max = 50.0;
1961 let mut scheduler = QueueingScheduler::new(config);
1962
1963 scheduler.submit_with_sources(
1964 20.0,
1965 1.0,
1966 WeightSource::Explicit,
1967 EstimateSource::Explicit,
1968 None::<&str>,
1969 );
1970
1971 let next = scheduler.peek_next().unwrap();
1972 assert!((next.weight - 20.0).abs() < f64::EPSILON);
1973 }
1974
1975 #[test]
1976 fn explicit_estimate_honors_config_p_max_above_defaults() {
1977 let mut config = test_config();
1978 config.p_max_ms = 100_000.0;
1979 let mut scheduler = QueueingScheduler::new(config);
1980
1981 scheduler.submit_with_sources(
1982 1.0,
1983 50_000.0,
1984 WeightSource::Explicit,
1985 EstimateSource::Explicit,
1986 None::<&str>,
1987 );
1988
1989 let next = scheduler.peek_next().unwrap();
1990 assert!((next.remaining_time - 50_000.0).abs() < f64::EPSILON);
1991 }
1992
1993 #[test]
1998 fn cancel_removes_job() {
1999 let mut scheduler = QueueingScheduler::new(test_config());
2000 let id = scheduler.submit(1.0, 10.0).unwrap();
2001
2002 assert!(scheduler.cancel(id));
2003 assert!(scheduler.peek_next().is_none());
2004 }
2005
2006 #[test]
2007 fn cancel_returns_false_for_nonexistent() {
2008 let mut scheduler = QueueingScheduler::new(test_config());
2009 assert!(!scheduler.cancel(999));
2010 }
2011
2012 #[test]
2017 fn reset_clears_all_state() {
2018 let mut scheduler = QueueingScheduler::new(test_config());
2019
2020 scheduler.submit(1.0, 10.0);
2021 scheduler.tick(5.0);
2022
2023 scheduler.reset();
2024
2025 assert!(scheduler.peek_next().is_none());
2026 assert_eq!(scheduler.stats().total_submitted, 0);
2027 assert_eq!(scheduler.stats().total_completed, 0);
2028 }
2029
2030 #[test]
2031 fn clear_removes_jobs_but_keeps_stats() {
2032 let mut scheduler = QueueingScheduler::new(test_config());
2033
2034 scheduler.submit(1.0, 10.0);
2035 scheduler.clear();
2036
2037 assert!(scheduler.peek_next().is_none());
2038 assert_eq!(scheduler.stats().total_submitted, 1); }
2040
2041 #[test]
2046 fn job_progress_increases() {
2047 let mut job = Job::new(1, 1.0, 100.0);
2048 assert_eq!(job.progress(), 0.0);
2049
2050 job.remaining_time = 50.0;
2051 assert!((job.progress() - 0.5).abs() < 0.01);
2052
2053 job.remaining_time = 0.0;
2054 assert_eq!(job.progress(), 1.0);
2055 }
2056
2057 #[test]
2058 fn job_is_complete() {
2059 let mut job = Job::new(1, 1.0, 10.0);
2060 assert!(!job.is_complete());
2061
2062 job.remaining_time = 0.0;
2063 assert!(job.is_complete());
2064 }
2065
2066 #[test]
2071 fn property_work_conserving() {
2072 let mut scheduler = QueueingScheduler::new(test_config());
2073
2074 for i in 0..10 {
2076 scheduler.submit(1.0, (i as f64) + 1.0);
2077 }
2078
2079 let mut total_processed = 0;
2081 while scheduler.peek_next().is_some() {
2082 let completed = scheduler.tick(1.0);
2083 total_processed += completed.len();
2084 }
2085
2086 assert_eq!(total_processed, 10);
2087 }
2088
2089 #[test]
2090 fn property_bounded_memory() {
2091 let mut config = test_config();
2092 config.max_queue_size = 100;
2093 let mut scheduler = QueueingScheduler::new(config);
2094
2095 for _ in 0..1000 {
2097 scheduler.submit(1.0, 10.0);
2098 }
2099
2100 assert!(scheduler.stats().queue_length <= 100);
2101 }
2102
2103 #[test]
2104 fn property_deterministic() {
2105 let run = || {
2106 let mut scheduler = QueueingScheduler::new(test_config());
2107 let mut completions = Vec::new();
2108
2109 for i in 0..20 {
2110 scheduler.submit(((i % 3) + 1) as f64, ((i % 5) + 1) as f64);
2111 }
2112
2113 for _ in 0..50 {
2114 completions.extend(scheduler.tick(1.0));
2115 }
2116
2117 completions
2118 };
2119
2120 let run1 = run();
2121 let run2 = run();
2122
2123 assert_eq!(run1, run2, "Scheduling should be deterministic");
2124 }
2125
2126 #[test]
2127 fn smith_beats_fifo_on_mixed_workload() {
2128 let workload = mixed_workload();
2129 let smith = simulate_policy(SimPolicy::Smith, &workload);
2130 let fifo = simulate_policy(SimPolicy::Fifo, &workload);
2131
2132 eprintln!("{}", workload_summary_json(&workload));
2133 eprintln!("{}", summary_json(SimPolicy::Smith, &smith));
2134 eprintln!("{}", summary_json(SimPolicy::Fifo, &fifo));
2135
2136 assert!(
2137 smith.mean < fifo.mean,
2138 "mean should improve: smith={} fifo={}",
2139 summary_json(SimPolicy::Smith, &smith),
2140 summary_json(SimPolicy::Fifo, &fifo)
2141 );
2142 assert!(
2143 smith.p95 < fifo.p95,
2144 "p95 should improve: smith={} fifo={}",
2145 summary_json(SimPolicy::Smith, &smith),
2146 summary_json(SimPolicy::Fifo, &fifo)
2147 );
2148 assert!(
2149 smith.p99 < fifo.p99,
2150 "p99 should improve: smith={} fifo={}",
2151 summary_json(SimPolicy::Smith, &smith),
2152 summary_json(SimPolicy::Fifo, &fifo)
2153 );
2154 }
2155
2156 #[test]
2157 fn simulation_is_deterministic_per_policy() {
2158 let workload = mixed_workload();
2159 let smith1 = simulate_policy(SimPolicy::Smith, &workload);
2160 let smith2 = simulate_policy(SimPolicy::Smith, &workload);
2161 let fifo1 = simulate_policy(SimPolicy::Fifo, &workload);
2162 let fifo2 = simulate_policy(SimPolicy::Fifo, &workload);
2163
2164 assert_eq!(smith1.completion_order, smith2.completion_order);
2165 assert_eq!(fifo1.completion_order, fifo2.completion_order);
2166 assert!((smith1.mean - smith2.mean).abs() < 1e-9);
2167 assert!((fifo1.mean - fifo2.mean).abs() < 1e-9);
2168 }
2169
2170 #[test]
2171 fn effect_queue_trace_is_deterministic() {
2172 let mut config = test_config();
2173 config.preemptive = false;
2174 config.aging_factor = 0.0;
2175 config.wait_starve_ms = 0.0;
2176 config.force_fifo = false;
2177 config.smith_enabled = true;
2178
2179 let mut scheduler = QueueingScheduler::new(config);
2180 let id_alpha = scheduler
2181 .submit_with_sources(
2182 1.0,
2183 8.0,
2184 WeightSource::Explicit,
2185 EstimateSource::Explicit,
2186 Some("alpha"),
2187 )
2188 .expect("alpha accepted");
2189 let id_beta = scheduler
2190 .submit_with_sources(
2191 4.0,
2192 2.0,
2193 WeightSource::Explicit,
2194 EstimateSource::Explicit,
2195 Some("beta"),
2196 )
2197 .expect("beta accepted");
2198 let id_gamma = scheduler
2199 .submit_with_sources(
2200 2.0,
2201 10.0,
2202 WeightSource::Explicit,
2203 EstimateSource::Explicit,
2204 Some("gamma"),
2205 )
2206 .expect("gamma accepted");
2207 let id_delta = scheduler
2208 .submit_with_sources(
2209 3.0,
2210 3.0,
2211 WeightSource::Explicit,
2212 EstimateSource::Explicit,
2213 Some("delta"),
2214 )
2215 .expect("delta accepted");
2216
2217 scheduler.refresh_priorities();
2218
2219 let mut selected = Vec::new();
2220 while let Some(job) = scheduler.peek_next().cloned() {
2221 let evidence = scheduler.evidence();
2222 if let Some(id) = evidence.selected_job_id {
2223 selected.push(id);
2224 }
2225 println!("{}", evidence.to_jsonl("effect_queue_select"));
2226
2227 let completed = scheduler.tick(job.remaining_time);
2228 assert!(
2229 !completed.is_empty(),
2230 "expected completion per tick in non-preemptive mode"
2231 );
2232 }
2233
2234 assert_eq!(selected, vec![id_beta, id_delta, id_gamma, id_alpha]);
2235 }
2236
2237 #[test]
2242 fn zero_weight_handled() {
2243 let mut scheduler = QueueingScheduler::new(test_config());
2244 scheduler.submit(0.0, 10.0);
2245 assert!(scheduler.peek_next().is_some());
2246 }
2247
2248 #[test]
2249 fn zero_time_completes_immediately() {
2250 let mut scheduler = QueueingScheduler::new(test_config());
2251 scheduler.submit(1.0, 0.0);
2252 let completed = scheduler.tick(1.0);
2253 assert_eq!(completed.len(), 1);
2254 }
2255
2256 #[test]
2257 fn negative_time_handled() {
2258 let mut scheduler = QueueingScheduler::new(test_config());
2259 scheduler.submit(1.0, -10.0);
2260 let completed = scheduler.tick(1.0);
2261 assert_eq!(completed.len(), 1);
2262 }
2263
2264 #[test]
2265 fn tick_non_finite_delta_noops() {
2266 let mut scheduler = QueueingScheduler::new(test_config());
2267 scheduler.submit(1.0, 5.0);
2268
2269 let before = scheduler.stats();
2270 assert!(scheduler.tick(f64::NAN).is_empty());
2271 assert!(scheduler.tick(f64::INFINITY).is_empty());
2272 assert!(scheduler.tick(f64::NEG_INFINITY).is_empty());
2273 let after = scheduler.stats();
2274
2275 assert_eq!(before.total_processing_time, after.total_processing_time);
2276 assert_eq!(before.total_completed, after.total_completed);
2277 assert!(scheduler.peek_next().is_some());
2278 }
2279
2280 #[test]
2285 fn job_new_nan_weight_clamps_to_min() {
2286 let job = Job::new(1, f64::NAN, 10.0);
2287 assert_eq!(job.weight, DEFAULT_W_MIN);
2288 }
2289
2290 #[test]
2291 fn job_new_pos_inf_weight_clamps_to_max() {
2292 let job = Job::new(1, f64::INFINITY, 10.0);
2293 assert_eq!(job.weight, DEFAULT_W_MAX);
2294 }
2295
2296 #[test]
2297 fn job_new_neg_inf_weight_clamps_to_min() {
2298 let job = Job::new(1, f64::NEG_INFINITY, 10.0);
2299 assert_eq!(job.weight, DEFAULT_W_MIN);
2300 }
2301
2302 #[test]
2303 fn job_new_nan_estimate_clamps_to_max() {
2304 let job = Job::new(1, 1.0, f64::NAN);
2305 assert_eq!(job.remaining_time, DEFAULT_P_MAX_MS);
2306 assert_eq!(job.total_time, DEFAULT_P_MAX_MS);
2307 }
2308
2309 #[test]
2310 fn job_new_pos_inf_estimate_clamps_to_max() {
2311 let job = Job::new(1, 1.0, f64::INFINITY);
2312 assert_eq!(job.remaining_time, DEFAULT_P_MAX_MS);
2313 }
2314
2315 #[test]
2316 fn job_new_neg_inf_estimate_clamps_to_min() {
2317 let job = Job::new(1, 1.0, f64::NEG_INFINITY);
2318 assert_eq!(job.remaining_time, DEFAULT_P_MIN_MS);
2319 }
2320
2321 #[test]
2322 fn job_with_name_sets_name() {
2323 let job = Job::with_name(1, 1.0, 10.0, "alpha");
2324 assert_eq!(job.name.as_deref(), Some("alpha"));
2325 assert_eq!(job.id, 1);
2326 }
2327
2328 #[test]
2329 fn job_with_sources_sets_both() {
2330 let job =
2331 Job::new(1, 1.0, 10.0).with_sources(WeightSource::Unknown, EstimateSource::Historical);
2332 assert_eq!(job.weight_source, WeightSource::Unknown);
2333 assert_eq!(job.estimate_source, EstimateSource::Historical);
2334 }
2335
2336 #[test]
2337 fn job_progress_zero_total_time() {
2338 let mut job = Job::new(1, 1.0, 10.0);
2339 job.total_time = 0.0;
2340 assert_eq!(job.progress(), 1.0);
2341 }
2342
2343 #[test]
2344 fn job_is_complete_negative_remaining() {
2345 let mut job = Job::new(1, 1.0, 10.0);
2346 job.remaining_time = -5.0;
2347 assert!(job.is_complete());
2348 }
2349
2350 #[test]
2355 fn submit_nan_weight_normalized() {
2356 let mut scheduler = QueueingScheduler::new(test_config());
2357 scheduler.submit(f64::NAN, 10.0);
2358 let next = scheduler.peek_next().unwrap();
2359 assert!(next.weight >= DEFAULT_W_MIN);
2360 assert!(next.weight.is_finite());
2361 }
2362
2363 #[test]
2364 fn submit_inf_weight_normalized() {
2365 let mut scheduler = QueueingScheduler::new(test_config());
2366 scheduler.submit(f64::INFINITY, 10.0);
2367 let next = scheduler.peek_next().unwrap();
2368 assert!(next.weight <= DEFAULT_W_MAX);
2369 assert!(next.weight.is_finite());
2370 }
2371
2372 #[test]
2373 fn submit_nan_estimate_normalized() {
2374 let mut scheduler = QueueingScheduler::new(test_config());
2375 scheduler.submit(1.0, f64::NAN);
2376 let next = scheduler.peek_next().unwrap();
2377 assert!(next.remaining_time <= DEFAULT_P_MAX_MS);
2378 assert!(next.remaining_time.is_finite());
2379 }
2380
2381 #[test]
2382 fn submit_inf_estimate_normalized() {
2383 let mut scheduler = QueueingScheduler::new(test_config());
2384 scheduler.submit(1.0, f64::INFINITY);
2385 let next = scheduler.peek_next().unwrap();
2386 assert!(next.remaining_time <= DEFAULT_P_MAX_MS);
2387 assert!(next.remaining_time.is_finite());
2388 }
2389
2390 #[test]
2395 fn config_mode_smith() {
2396 let config = SchedulerConfig {
2397 smith_enabled: true,
2398 force_fifo: false,
2399 ..Default::default()
2400 };
2401 assert_eq!(config.mode(), SchedulingMode::Smith);
2402 }
2403
2404 #[test]
2405 fn config_mode_srpt() {
2406 let config = SchedulerConfig {
2407 smith_enabled: false,
2408 force_fifo: false,
2409 ..Default::default()
2410 };
2411 assert_eq!(config.mode(), SchedulingMode::Srpt);
2412 }
2413
2414 #[test]
2415 fn config_mode_fifo_overrides_smith() {
2416 let config = SchedulerConfig {
2417 smith_enabled: true,
2418 force_fifo: true,
2419 ..Default::default()
2420 };
2421 assert_eq!(config.mode(), SchedulingMode::Fifo);
2422 }
2423
2424 #[test]
2429 fn starvation_guard_triggers_after_threshold() {
2430 let mut config = test_config();
2431 config.aging_factor = 0.0;
2432 config.wait_starve_ms = 50.0;
2433 config.starve_boost_ratio = 5.0;
2434 let mut scheduler = QueueingScheduler::new(config);
2435
2436 scheduler.submit(1.0, 100.0); scheduler.current_time = 60.0; scheduler.refresh_priorities();
2439
2440 let evidence = scheduler.evidence();
2441 let job_ev = &evidence.jobs[0];
2442 assert!(
2444 job_ev.starvation_floor > 0.0,
2445 "starvation floor should be active: {}",
2446 job_ev.starvation_floor
2447 );
2448 assert!(
2449 job_ev.effective_priority >= job_ev.starvation_floor,
2450 "effective priority {} should be >= starvation floor {}",
2451 job_ev.effective_priority,
2452 job_ev.starvation_floor
2453 );
2454 }
2455
2456 #[test]
2457 fn starvation_guard_disabled_when_zero() {
2458 let mut config = test_config();
2459 config.aging_factor = 0.0;
2460 config.wait_starve_ms = 0.0;
2461 let mut scheduler = QueueingScheduler::new(config);
2462
2463 scheduler.submit(1.0, 100.0);
2464 scheduler.current_time = 1000.0;
2465 scheduler.refresh_priorities();
2466
2467 let evidence = scheduler.evidence();
2468 let job_ev = &evidence.jobs[0];
2469 assert!(
2470 (job_ev.starvation_floor - 0.0).abs() < f64::EPSILON,
2471 "starvation floor should be 0 when disabled"
2472 );
2473 }
2474
2475 #[test]
2480 fn cancel_current_job() {
2481 let mut scheduler = QueueingScheduler::new(test_config());
2482 let id = scheduler.submit(1.0, 100.0).unwrap();
2483 scheduler.tick(10.0); assert!(scheduler.cancel(id));
2486 assert!(scheduler.peek_next().is_none());
2487 }
2488
2489 #[test]
2490 fn cancel_from_middle_of_queue() {
2491 let mut scheduler = QueueingScheduler::new(test_config());
2492 scheduler.submit(1.0, 100.0); let id2 = scheduler.submit(1.0, 50.0).unwrap(); scheduler.submit(1.0, 200.0); assert!(scheduler.cancel(id2));
2497 assert_eq!(scheduler.stats().queue_length, 2);
2498 }
2499
2500 #[test]
2505 fn tick_negative_delta_returns_empty() {
2506 let mut scheduler = QueueingScheduler::new(test_config());
2507 scheduler.submit(1.0, 10.0);
2508 let completed = scheduler.tick(-5.0);
2509 assert!(completed.is_empty());
2510 }
2511
2512 #[test]
2513 fn tick_empty_queue_advances_time() {
2514 let mut scheduler = QueueingScheduler::new(test_config());
2515 let completed = scheduler.tick(100.0);
2516 assert!(completed.is_empty());
2517 }
2518
2519 #[test]
2520 fn tick_processes_across_multiple_jobs_in_single_delta() {
2521 let mut config = test_config();
2522 config.aging_factor = 0.0;
2523 let mut scheduler = QueueingScheduler::new(config);
2524
2525 scheduler.submit(1.0, 3.0);
2526 scheduler.submit(1.0, 3.0);
2527 scheduler.submit(1.0, 3.0);
2528
2529 let completed = scheduler.tick(9.0);
2531 assert_eq!(completed.len(), 3);
2532 }
2533
2534 #[test]
2539 fn stats_default_values() {
2540 let stats = SchedulerStats::default();
2541 assert_eq!(stats.total_submitted, 0);
2542 assert_eq!(stats.total_completed, 0);
2543 assert_eq!(stats.total_rejected, 0);
2544 assert_eq!(stats.total_preemptions, 0);
2545 assert_eq!(stats.queue_length, 0);
2546 }
2547
2548 #[test]
2549 fn stats_mean_response_time_zero_completions() {
2550 let stats = SchedulerStats::default();
2551 assert_eq!(stats.mean_response_time(), 0.0);
2552 }
2553
2554 #[test]
2555 fn stats_throughput_zero_processing_time() {
2556 let stats = SchedulerStats::default();
2557 assert_eq!(stats.throughput(), 0.0);
2558 }
2559
2560 #[test]
2561 fn stats_max_response_time_tracked() {
2562 let mut scheduler = QueueingScheduler::new(test_config());
2563 scheduler.submit(1.0, 5.0);
2564 scheduler.submit(1.0, 10.0);
2565 scheduler.tick(15.0);
2566
2567 let stats = scheduler.stats();
2568 assert!(
2569 stats.max_response_time >= 10.0,
2570 "max response time {} should be >= 10",
2571 stats.max_response_time
2572 );
2573 }
2574
2575 #[test]
2580 fn evidence_continuation_reason() {
2581 let mut scheduler = QueueingScheduler::new(test_config());
2582 scheduler.submit(1.0, 100.0);
2583 scheduler.tick(10.0); let evidence = scheduler.evidence();
2586 assert_eq!(evidence.reason, SelectionReason::Continuation);
2587 }
2588
2589 #[test]
2590 fn evidence_single_job_no_tie_break() {
2591 let mut scheduler = QueueingScheduler::new(test_config());
2592 scheduler.submit(1.0, 10.0);
2593
2594 let evidence = scheduler.evidence();
2595 assert!(
2596 evidence.tie_break_reason.is_none(),
2597 "single job should have no tie break"
2598 );
2599 }
2600
2601 #[test]
2602 fn evidence_to_jsonl_contains_required_fields() {
2603 let mut scheduler = QueueingScheduler::new(test_config());
2604 scheduler.submit(1.0, 10.0);
2605 scheduler.submit(2.0, 5.0);
2606
2607 let evidence = scheduler.evidence();
2608 let json = evidence.to_jsonl("test_event");
2609
2610 assert!(json.contains("\"event\":\"test_event\""));
2611 assert!(json.contains("\"current_time\":"));
2612 assert!(json.contains("\"selected_job_id\":"));
2613 assert!(json.contains("\"queue_length\":"));
2614 assert!(json.contains("\"mean_wait_time\":"));
2615 assert!(json.contains("\"max_wait_time\":"));
2616 assert!(json.contains("\"reason\":"));
2617 assert!(json.contains("\"tie_break_reason\":"));
2618 assert!(json.contains("\"jobs\":["));
2619 }
2620
2621 #[test]
2622 fn evidence_to_jsonl_empty_queue() {
2623 let scheduler = QueueingScheduler::new(test_config());
2624 let evidence = scheduler.evidence();
2625 let json = evidence.to_jsonl("empty");
2626
2627 assert!(json.contains("\"selected_job_id\":null"));
2628 assert!(json.contains("\"tie_break_reason\":null"));
2629 assert!(json.contains("\"jobs\":[]"));
2630 }
2631
2632 #[test]
2633 fn job_evidence_to_json_contains_all_fields() {
2634 let mut config = test_config();
2635 config.aging_factor = 0.5;
2636 config.wait_starve_ms = 5.0;
2637 let mut scheduler = QueueingScheduler::new(config);
2638
2639 scheduler.submit_with_sources(
2640 2.0,
2641 10.0,
2642 WeightSource::Explicit,
2643 EstimateSource::Default,
2644 Some("test-job"),
2645 );
2646 scheduler.current_time = 10.0;
2647 scheduler.refresh_priorities();
2648
2649 let evidence = scheduler.evidence();
2650 let json = evidence.to_jsonl("detail");
2651
2652 assert!(json.contains("\"job_id\":"));
2653 assert!(json.contains("\"name\":\"test-job\""));
2654 assert!(json.contains("\"estimate_ms\":"));
2655 assert!(json.contains("\"weight\":"));
2656 assert!(json.contains("\"ratio\":"));
2657 assert!(json.contains("\"aging_reward\":"));
2658 assert!(json.contains("\"starvation_floor\":"));
2659 assert!(json.contains("\"age_ms\":"));
2660 assert!(json.contains("\"effective_priority\":"));
2661 assert!(json.contains("\"objective_loss_proxy\":"));
2662 assert!(json.contains("\"estimate_source\":"));
2663 assert!(json.contains("\"weight_source\":"));
2664 }
2665
2666 #[test]
2667 fn evidence_jsonl_escapes_special_chars_in_name() {
2668 let mut scheduler = QueueingScheduler::new(test_config());
2669 scheduler.submit_named(1.0, 10.0, Some("job\"with\\special\nchars"));
2670
2671 let evidence = scheduler.evidence();
2672 let json = evidence.to_jsonl("escape_test");
2673
2674 assert!(json.contains("\\\""));
2675 assert!(json.contains("\\\\"));
2676 assert!(json.contains("\\n"));
2677 }
2678
2679 #[test]
2684 fn selection_reason_as_str_coverage() {
2685 assert_eq!(SelectionReason::QueueEmpty.as_str(), "queue_empty");
2686 assert_eq!(
2687 SelectionReason::ShortestRemaining.as_str(),
2688 "shortest_remaining"
2689 );
2690 assert_eq!(
2691 SelectionReason::HighestWeightedPriority.as_str(),
2692 "highest_weighted_priority"
2693 );
2694 assert_eq!(SelectionReason::Fifo.as_str(), "fifo");
2695 assert_eq!(SelectionReason::AgingBoost.as_str(), "aging_boost");
2696 assert_eq!(SelectionReason::Continuation.as_str(), "continuation");
2697 }
2698
2699 #[test]
2700 fn estimate_source_as_str_coverage() {
2701 assert_eq!(EstimateSource::Explicit.as_str(), "explicit");
2702 assert_eq!(EstimateSource::Historical.as_str(), "historical");
2703 assert_eq!(EstimateSource::Default.as_str(), "default");
2704 assert_eq!(EstimateSource::Unknown.as_str(), "unknown");
2705 }
2706
2707 #[test]
2708 fn weight_source_as_str_coverage() {
2709 assert_eq!(WeightSource::Explicit.as_str(), "explicit");
2710 assert_eq!(WeightSource::Default.as_str(), "default");
2711 assert_eq!(WeightSource::Unknown.as_str(), "unknown");
2712 }
2713
2714 #[test]
2715 fn tie_break_reason_as_str_coverage() {
2716 assert_eq!(
2717 TieBreakReason::EffectivePriority.as_str(),
2718 "effective_priority"
2719 );
2720 assert_eq!(TieBreakReason::BaseRatio.as_str(), "base_ratio");
2721 assert_eq!(TieBreakReason::Weight.as_str(), "weight");
2722 assert_eq!(TieBreakReason::RemainingTime.as_str(), "remaining_time");
2723 assert_eq!(TieBreakReason::ArrivalSeq.as_str(), "arrival_seq");
2724 assert_eq!(TieBreakReason::JobId.as_str(), "job_id");
2725 assert_eq!(TieBreakReason::Continuation.as_str(), "continuation");
2726 }
2727
2728 #[test]
2733 fn debug_job() {
2734 let job = Job::with_name(1, 2.0, 50.0, "render");
2735 let dbg = format!("{job:?}");
2736 assert!(dbg.contains("Job"));
2737 assert!(dbg.contains("render"));
2738 }
2739
2740 #[test]
2741 fn debug_scheduler_config() {
2742 let config = SchedulerConfig::default();
2743 let dbg = format!("{config:?}");
2744 assert!(dbg.contains("SchedulerConfig"));
2745 assert!(dbg.contains("aging_factor"));
2746 }
2747
2748 #[test]
2749 fn debug_scheduler_stats() {
2750 let stats = SchedulerStats::default();
2751 let dbg = format!("{stats:?}");
2752 assert!(dbg.contains("SchedulerStats"));
2753 }
2754
2755 #[test]
2756 fn debug_scheduling_evidence() {
2757 let scheduler = QueueingScheduler::new(test_config());
2758 let evidence = scheduler.evidence();
2759 let dbg = format!("{evidence:?}");
2760 assert!(dbg.contains("SchedulingEvidence"));
2761 }
2762
2763 #[test]
2764 fn debug_scheduling_mode() {
2765 assert!(format!("{:?}", SchedulingMode::Smith).contains("Smith"));
2766 assert!(format!("{:?}", SchedulingMode::Srpt).contains("Srpt"));
2767 assert!(format!("{:?}", SchedulingMode::Fifo).contains("Fifo"));
2768 }
2769
2770 #[test]
2775 fn historical_estimate_passes_through() {
2776 let mut scheduler = QueueingScheduler::new(test_config());
2777 scheduler.submit_with_sources(
2778 1.0,
2779 42.0,
2780 WeightSource::Explicit,
2781 EstimateSource::Historical,
2782 None::<&str>,
2783 );
2784
2785 let next = scheduler.peek_next().unwrap();
2786 assert!((next.remaining_time - 42.0).abs() < f64::EPSILON);
2787 }
2788
2789 #[test]
2794 fn multiple_preemptions_counted() {
2795 let mut config = test_config();
2796 config.aging_factor = 0.0; config.wait_starve_ms = 0.0;
2798 let mut scheduler = QueueingScheduler::new(config);
2799
2800 scheduler.submit(1.0, 100.0); scheduler.tick(1.0); scheduler.submit(1.0, 50.0); scheduler.tick(1.0); scheduler.submit(1.0, 10.0); assert!(
2809 scheduler.stats().total_preemptions >= 2,
2810 "expected >= 2 preemptions, got {}",
2811 scheduler.stats().total_preemptions
2812 );
2813 }
2814
2815 #[test]
2820 fn multiple_rejections_counted() {
2821 let mut config = test_config();
2822 config.max_queue_size = 1;
2823 let mut scheduler = QueueingScheduler::new(config);
2824
2825 scheduler.submit(1.0, 10.0); scheduler.submit(1.0, 10.0); scheduler.submit(1.0, 10.0); assert_eq!(scheduler.stats().total_rejected, 2);
2830 }
2831
2832 #[test]
2837 fn reset_resets_job_id_sequence() {
2838 let mut scheduler = QueueingScheduler::new(test_config());
2839 scheduler.submit(1.0, 10.0); scheduler.submit(1.0, 10.0); scheduler.reset();
2843
2844 let id = scheduler.submit(1.0, 10.0).unwrap();
2845 assert_eq!(id, 1, "job id should restart from 1 after reset");
2846 }
2847
2848 #[test]
2849 fn clear_preserves_job_id_sequence() {
2850 let mut scheduler = QueueingScheduler::new(test_config());
2851 scheduler.submit(1.0, 10.0); scheduler.submit(1.0, 10.0); scheduler.clear();
2855
2856 let id = scheduler.submit(1.0, 10.0).unwrap();
2857 assert_eq!(id, 3, "job id should continue after clear");
2858 }
2859
2860 #[test]
2865 fn evidence_aging_boost_reason() {
2866 let mut config = test_config();
2867 config.aging_factor = 1.0;
2868 config.wait_starve_ms = 10.0;
2869 let mut scheduler = QueueingScheduler::new(config);
2870
2871 scheduler.submit(1.0, 100.0);
2872 scheduler.current_time = 100.0;
2873 scheduler.refresh_priorities();
2874
2875 let evidence = scheduler.evidence();
2876 assert_eq!(
2877 evidence.reason,
2878 SelectionReason::AgingBoost,
2879 "long-waiting job should show aging boost reason"
2880 );
2881 }
2882}