// This Source Code Form is subject to the terms of the Mozilla Public // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at https://mozilla.org/MPL/2.0/.
usecrate::metrics::{DatetimeMetric, StringMetric, TimeUnit}; usecrate::storage::INTERNAL_STORAGE; usecrate::util::local_now_with_offset; usecrate::{CommonMetricData, Glean, Lifetime}; use chrono::prelude::*; use chrono::Days; use once_cell::sync::Lazy; use std::sync::{Arc, Condvar, Mutex}; use std::thread::JoinHandle;
const SCHEDULED_HOUR: u32 = 4;
// Clippy thinks an AtomicBool would be preferred, but Condvar requires a full Mutex. // See https://github.com/rust-lang/rust-clippy/issues/1516 #[allow(clippy::mutex_atomic)] static TASK_CONDVAR: Lazy<Arc<(Mutex<bool>, Condvar)>> =
Lazy::new(|| Arc::new((Mutex::new(false), Condvar::new())));
/// Describes the interface for a submitter of "metrics" pings. /// Used to decouple the implementation so we can test it. trait MetricsPingSubmitter { /// Submits a metrics ping, updating the last sent time to `now` /// (which might not be _right now_ due to processing delays (or in tests)) fn submit_metrics_ping(&self, glean: &Glean, reason: Option<&str>, now: DateTime<FixedOffset>);
}
/// Describes the interface for a scheduler of "metrics" pings. /// Used to decouple the implementation so we can test it. trait MetricsPingScheduler { /// Begins a recurring schedule of "metrics" ping submissions, on another thread. /// `now` is used with `when` to determine the first schedule interval and /// may not be _right now_ due to processing delays (or in tests). fn start_scheduler(
&self,
submitter: impl MetricsPingSubmitter + Send + 'static,
now: DateTime<FixedOffset>,
when: When,
);
}
/// Uses Glean to submit "metrics" pings directly. struct GleanMetricsPingSubmitter {} impl MetricsPingSubmitter for GleanMetricsPingSubmitter { fn submit_metrics_ping(&self, glean: &Glean, reason: Option<&str>, now: DateTime<FixedOffset>) {
glean.submit_ping_by_name("metrics", reason); // Always update the collection date, irrespective of the ping being sent.
get_last_sent_time_metric().set_sync_chrono(glean, now);
}
}
/// Performs startup checks to decide when to schedule the next "metrics" ping collection. /// **Must** be called before draining the preinit queue. /// (We're at the Language Bindings' mercy for that) pubfn schedule(glean: &Glean) { let now = local_now_with_offset();
let (cancelled_lock, _condvar) = &**TASK_CONDVAR; if *cancelled_lock.lock().unwrap() {
log::debug!("Told to schedule, but already cancelled. Are we in a test?");
}
*cancelled_lock.lock().unwrap() = false; // Uncancel the thread.
let submitter = GleanMetricsPingSubmitter {}; let scheduler = GleanMetricsPingScheduler {};
/// Tells the scheduler task to exit quickly and cleanly. pubfn cancel() { let (cancelled_lock, condvar) = &**TASK_CONDVAR; // One `*` for Lazy, the second for Arc
*cancelled_lock.lock().unwrap() = true; // Cancel the scheduler thread.
condvar.notify_all(); // Notify any/all listening schedulers to check whether they were cancelled.
}
fn schedule_internal(
glean: &Glean,
submitter: impl MetricsPingSubmitter + Send + 'static,
scheduler: impl MetricsPingScheduler,
now: DateTime<FixedOffset>,
) { let last_sent_build_metric = get_last_sent_build_metric(); iflet Some(last_sent_build) = last_sent_build_metric.get_value(glean, Some(INTERNAL_STORAGE)) { // If `app_build` is longer than StringMetric's max length, we will always // treat it as a changed build when really it isn't. // This will be externally-observable as InvalidOverflow errors on both the core // `client_info.app_build` metric and the scheduler's internal metric. if last_sent_build != glean.app_build {
last_sent_build_metric.set_sync(glean, &glean.app_build);
log::info!("App build changed. Sending 'metrics' ping");
submitter.submit_metrics_ping(glean, Some("upgrade"), now);
scheduler.start_scheduler(submitter, now, When::Reschedule); return;
}
} else { // No value in last_sent_build. Better set one.
last_sent_build_metric.set_sync(glean, &glean.app_build);
}
let last_sent_time = get_last_sent_time_metric().get_value(glean, INTERNAL_STORAGE); iflet Some(last_sent) = last_sent_time {
log::info!("The 'metrics' ping was last sent on {}", last_sent);
}
// We aim to cover 3 cases here: // // 1. The ping was already collected on the current calendar day; // only schedule one for collection on the next calendar day at the due time. // 2. The ping was NOT collected on the current calendar day AND we're later // than today's due time; collect the ping immediately. // 3. The ping was NOT collected on the current calendar day BUT we still have // some time to the due time; schedule for submitting the current calendar day.
let already_sent_today = last_sent_time.is_some_and(|d| d.date_naive() == now.date_naive()); // Today's 04:00 in local time let cutoff_time = now
.naive_local()
.date()
.and_hms_opt(SCHEDULED_HOUR, 0, 0)
.unwrap()
.and_local_timezone(now.timezone())
.unwrap();
if already_sent_today { // Case #1
log::info!("The 'metrics' ping was already sent today, {}", now);
scheduler.start_scheduler(submitter, now, When::Tomorrow);
} elseif now > cutoff_time { // Case #2
log::info!("Sending the 'metrics' ping immediately, {}", now);
submitter.submit_metrics_ping(glean, Some("overdue"), now);
scheduler.start_scheduler(submitter, now, When::Reschedule);
} else { // Case #3
log::info!("The 'metrics' collection is scheduled for today, {}", now);
scheduler.start_scheduler(submitter, now, When::Today);
}
}
impl When { /// Returns the duration from now until our deadline. /// Note that std::time::Duration doesn't do negative time spans, so if /// our deadline has passed, this will return zero. fn until(&self, now: DateTime<FixedOffset>) -> std::time::Duration { let now_local = now.naive_local();
let fire_date = matchself { Self::Today => now_local.date().and_hms_opt(SCHEDULED_HOUR, 0, 0).unwrap(), // Doesn't actually save us from being an hour off on DST because // chrono doesn't know when DST changes. : ( Self::Tomorrow | Self::Reschedule => { let next_day = now_local.checked_add_days(Days::new(1)).unwrap(); let next_day_date = next_day.date();
next_day_date.and_hms_opt(SCHEDULED_HOUR, 0, 0).unwrap()
}
};
(fire_date - now_local).to_std().unwrap_or_else(|_| { // If we're somehow out of range schedule 24 hours into the future. // We do NOT want to schedule a ping submission immediately.
std::time::Duration::from_secs(24 * 60 * 60)
})
}
fn start_scheduler(
submitter: impl MetricsPingSubmitter + Send + 'static,
now: DateTime<FixedOffset>,
when: When,
) -> JoinHandle<()> { let pair = Arc::clone(&TASK_CONDVAR); crate::thread::spawn("glean.mps", move || { let (cancelled_lock, condvar) = &*pair; letmut when = when; letmut now = now; loop { let dur = when.until(now);
log::info!("Scheduling for {} after {:?}, reason {:?}", now, dur, when); letmut timed_out = false;
{ match condvar.wait_timeout_while(cancelled_lock.lock().unwrap(), dur, |cancelled| {
!*cancelled
}) {
Err(err) => {
log::warn!("Condvar wait failure. MPS exiting. {}", err); break;
}
Ok((cancelled, wait_result)) => { if *cancelled {
log::info!("Metrics Ping Scheduler cancelled. Exiting."); break;
} elseif wait_result.timed_out() { // Can't get the global glean while holding cancelled's lock.
timed_out = true;
} else { // This should be impossible. `cancelled_lock` is acquired, and // `!*cancelled` is checked by the condvar before it is allowed // to return from `wait_timeout_while` (I checked). // So `Ok(_)` implies `*cancelled || wait_result.timed_out`.
log::warn!("Spurious wakeup of the MPS condvar should be impossible.");
}
}
}
} // Safety: // We are okay dropping the condvar's cancelled lock here because it only guards // whether we're cancelled, and we've established that we weren't when we timed out. // We might _now_ be cancelled at any time, in which case when we loop back over // we'll immediately exit. But first we need to submit our "metrics" ping. if timed_out {
log::info!("Time to submit our metrics ping, {:?}", when); let glean = crate::core::global_glean()
.expect( "Global Glean not present when trying to send scheduled 'metrics' ping?!",
)
.lock()
.unwrap();
submitter.submit_metrics_ping(&glean, Some(when.reason()), now);
when = When::Reschedule;
}
now = local_now_with_offset();
}
})
.expect("Unable to spawn Metrics Ping Scheduler thread.")
}
// Ensure on first run that we actually set the last sent build metric. // (and that we send an "overdue" ping if it's after the scheduled hour) #[test] fn first_run_last_sent_build() { let (mut glean, _t) = new_glean(None);
glean.app_build = "a build".into(); let lsb_metric = get_last_sent_build_metric();
assert_eq!(None, lsb_metric.get_value(&glean, Some(INTERNAL_STORAGE)));
// Ensure that if we have a different build, we immediately submit an "upgrade" ping // and schedule a "reschedule" ping for tomorrow. #[test] fn different_app_builds_submit_and_reschedule() { let (mut glean, _t) = new_glean(None);
glean.app_build = "a build".into();
get_last_sent_build_metric().set_sync(&glean, "a different build");
// If we've already sent a ping today, ensure we don't send a ping but we // do schedule a ping for tomorrow. ("Case #1" in schedule_internal) #[test] fn case_1_no_submit_but_schedule_tomorrow() { let (glean, _t) = new_glean(None);
// If we haven't sent a ping today and we're after the scheduled time, // ensure we send a ping and then schedule a "reschedule" ping for tomorrow. // ("Case #2" in schedule_internal) #[test] fn case_2_submit_ping_and_reschedule() { let (glean, _t) = new_glean(None);
let fake_yesterday = FixedOffset::east_opt(0)
.unwrap()
.with_ymd_and_hms(2021, 4, 29, SCHEDULED_HOUR, 0, 1)
.unwrap();
get_last_sent_time_metric().set_sync_chrono(&glean, fake_yesterday); let fake_now = fake_yesterday + Duration::days(1);
// If we haven't sent a ping today and we're before the scheduled time, // ensure we don't send a ping but schedule a "today" ping for today. // ("Case #3" in schedule_internal) #[test] fn case_3_no_submit_but_schedule_today() { let (glean, _t) = new_glean(None);
// `When` is responsible for date math. Let's make sure it's correct. #[test] fn when_gets_at_least_some_date_math_correct() { let now = FixedOffset::east_opt(0)
.unwrap()
.with_ymd_and_hms(2021, 4, 30, 15, 2, 10)
.unwrap(); // `now` is after `SCHEDULED_HOUR` but we should never schedule immediately:
assert_ne!(std::time::Duration::from_secs(0), When::Today.until(now)); // If we bring it back before `SCHEDULED_HOUR` it should give us the duration: let earlier = now
.date_naive()
.and_hms_opt(SCHEDULED_HOUR - 1, 0, 0)
.unwrap();
assert_eq!(
std::time::Duration::from_secs(3600),
When::Today.until(Utc.from_utc_datetime(&earlier).into())
);
// `Tomorrow` and `Reschedule` should differ only in their `reason()` // 46670s is 12h57m10s (aka, the time from 15:02:10 to 04:00:00 // (when the timezone doesn't change between them)).
assert_eq!(
std::time::Duration::from_secs(46670),
When::Tomorrow.until(now)
);
assert_eq!(
std::time::Duration::from_secs(46670),
When::Reschedule.until(now)
);
assert_eq!(When::Tomorrow.until(now), When::Reschedule.until(now));
assert_ne!(When::Tomorrow.reason(), When::Reschedule.reason());
}
#[test] fn datetime_offset_doesnt_cause_rapid_rescheduling() { let now = FixedOffset::west_opt(3600 * 7)
.unwrap()
.with_ymd_and_hms(2025, 7, 27, 22, 27, 59)
.unwrap();
let next_schedule = When::Reschedule.until(now);
// 22:27:59 -> (next day) 04:00 is 5h 32min 1s = 19921 seconds let expected_duration = std::time::Duration::from_secs(19921);
assert_eq!(expected_duration, next_schedule);
}
#[test] fn todays_scheduling_is_in_localtime() { let now = FixedOffset::west_opt(3600 * 7)
.unwrap()
.with_ymd_and_hms(2025, 7, 27, 3, 30, 0)
.unwrap();
let next_schedule = When::Today.until(now);
// 03:30:00 -> 04:00 is 30min let expected_duration = std::time::Duration::from_secs(30 * 60);
assert_eq!(expected_duration, next_schedule);
}
// Scheduler tests mutate global state and thus must not be run in parallel. // Otherwise one test could cancel the other. // This Mutex aims to solve that. static SCHEDULER_TEST_MUTEX: Lazy<Mutex<()>> = Lazy::new(|| Mutex::new(()));
// The scheduler has been designed to be cancellable. Can we cancel it? #[test] fn cancellable_tasks_can_be_cancelled() { // First and foremost, all scheduler tests must ensure they start uncancelled. // Perils of having shared state. let _test_lock = SCHEDULER_TEST_MUTEX.lock().unwrap(); let (cancelled_lock, _condvar) = &**TASK_CONDVAR; // One `*` for Lazy, the second for Arc
*cancelled_lock.lock().unwrap() = false;
// Pick a time at least two hours from the next scheduled submission. // (So that this test will time out if cancellation fails). let now = FixedOffset::east_opt(0)
.unwrap()
.with_ymd_and_hms(2021, 4, 30, SCHEDULED_HOUR - 2, 0, 0)
.unwrap();
let proxy_factory = || {
new_proxies(
|_, reason| {
panic!( "Shouldn't submit when testing scheduler. reason: {:?}",
reason
)
},
|_, _| panic!("Not even using the scheduler this time."),
)
};
// Test Today. let (submitter, submitter_count, _, _) = proxy_factory(); let handle = start_scheduler(submitter, now, When::Today); super::cancel();
handle.join().unwrap(); // Should complete immediately.
assert_eq!(0, submitter_count.swap(0, Ordering::Relaxed));
// Test Tomorrow. let (submitter, submitter_count, _, _) = proxy_factory();
*cancelled_lock.lock().unwrap() = false; // Uncancel. let handle = start_scheduler(submitter, now, When::Tomorrow); super::cancel();
handle.join().unwrap(); // Should complete immediately.
assert_eq!(0, submitter_count.swap(0, Ordering::Relaxed));
// Test Reschedule. let (submitter, submitter_count, _, _) = proxy_factory();
*cancelled_lock.lock().unwrap() = false; // Uncancel. let handle = start_scheduler(submitter, now, When::Reschedule); super::cancel();
handle.join().unwrap(); // Should complete immediately.
assert_eq!(0, submitter_count.swap(0, Ordering::Relaxed));
}
// We're not keen to wait like the scheduler is, but we can test a quick schedule. #[test] fn immediate_task_runs_immediately() { let _ = env_logger::builder().try_init(); // First and foremost, all scheduler tests must ensure they start uncancelled. // Perils of having shared state. let _test_lock = SCHEDULER_TEST_MUTEX.lock().unwrap(); let (cancelled_lock, _condvar) = &**TASK_CONDVAR; // One `*` for Lazy, the second for Arc
*cancelled_lock.lock().unwrap() = false;
// We're actually going to submit a ping from the scheduler, which requires a global glean. let (glean, _t) = new_glean(None);
assert!(
!glean.schedule_metrics_pings, "Real schedulers not allowed in tests!"
);
assert!(crate::core::setup_glean(glean).is_ok());
// We're choosing the exact `SCHEDULED_HOUR` to give us a duration of 0. let now = FixedOffset::east_opt(0)
.unwrap()
.with_ymd_and_hms(2021, 4, 21, 4, 0, 0)
.unwrap();
let (submitter, submitter_count, _, _) = new_proxies( move |_, reason| {
assert_eq!(reason, Some("today")); // After submitting the ping we expect, let's cancel this scheduler so the thread exits. // (But do it on another thread because the condvar loop is currently holding `cancelled`'s mutex)
std::thread::spawn(super::cancel);
},
|_, _| panic!("Not using the scheduler this time."),
);
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.