diff --git a/pipeline-engines/src/data_update_service.rs b/pipeline-engines/src/data_update_service.rs index 6c6e83d..b0ebadd 100644 --- a/pipeline-engines/src/data_update_service.rs +++ b/pipeline-engines/src/data_update_service.rs @@ -55,7 +55,7 @@ //! and a `429` response surfaces the server's `Retry-After` in the error but the //! next poll still uses the configured interval rather than that delay. -use std::collections::hash_map::DefaultHasher; +use std::collections::hash_map::RandomState; use std::hash::{Hash, Hasher}; use std::io::Read; use std::path::Path; @@ -279,30 +279,32 @@ impl DataUpdateService { .unwrap_or_else(|_| Duration::from_secs(0)); return Instant::now() + wait - + self.randomisation(data_file.configuration().max_randomisation_seconds); + + Self::randomisation(data_file.configuration().max_randomisation_seconds); } } - Instant::now() + self.poll_interval(data_file) + Instant::now() + Self::poll_interval(data_file) } - /// The base polling interval for a file plus randomisation. - fn poll_interval(&self, data_file: &AspectEngineDataFile) -> Duration { + /// The polling interval for a file plus a random delay of up to its + /// maximum randomisation, as each poll waits. + fn poll_interval(data_file: &AspectEngineDataFile) -> Duration { let config = data_file.configuration(); Duration::from_secs(config.polling_interval_seconds) - + self.randomisation(config.max_randomisation_seconds) + + Self::randomisation(config.max_randomisation_seconds) } - /// A pseudo-random extra delay between zero and `max_seconds`. + /// A random delay between zero and `max_seconds`, so instances that start + /// together poll at different times. /// - /// A hash of the current time seeds the choice, which is enough to stagger - /// many instances without pulling in a random-number-generator dependency. - fn randomisation(&self, max_seconds: u64) -> Duration { + /// Each `RandomState` has its own keys, which the standard library draws + /// from the operating system, so the delay differs between calls and + /// between processes without a random number generator dependency. + fn randomisation(max_seconds: u64) -> Duration { if max_seconds == 0 { return Duration::from_secs(0); } - let mut hasher = DefaultHasher::new(); - Instant::now().elapsed().as_nanos().hash(&mut hasher); - std::thread::current().id().hash(&mut hasher); + let mut hasher = std::hash::BuildHasher::build_hasher(&RandomState::new()); + std::time::SystemTime::now().hash(&mut hasher); Duration::from_secs(hasher.finish() % max_seconds) } @@ -513,12 +515,7 @@ impl ServiceHandle { // Poll now. A failure is logged, and the next interval retries. let result = self.check_for_update_from_url(®istration); log_poll(®istration, result); - let interval = Duration::from_secs( - registration - .data_file - .configuration() - .polling_interval_seconds, - ); + let interval = DataUpdateService::poll_interval(®istration.data_file); let next = now + interval; *registration .next_poll @@ -819,4 +816,140 @@ mod tests { let service = DataUpdateService::new(); assert_eq!(service.registered_count(), 0); } + + /// An engine with one data file and no update URL, so a poll of it makes + /// no request. + struct IdleEngine { + filter: fiftyone_pipeline_core::EvidenceKeyFilterWhitelist, + data_files: Vec>, + } + + impl fiftyone_pipeline_core::FlowElement for IdleEngine { + fn process( + &self, + _data: &mut fiftyone_pipeline_core::FlowData, + ) -> fiftyone_pipeline_core::Result<()> { + Ok(()) + } + fn data_key(&self) -> &str { + "idle" + } + fn evidence_key_filter(&self) -> &dyn fiftyone_pipeline_core::EvidenceKeyFilter { + &self.filter + } + fn properties(&self) -> &[fiftyone_pipeline_core::PropertyMetaData] { + &[] + } + } + + impl crate::AspectEngine for IdleEngine { + fn data_source_tier(&self) -> &str { + "Lite" + } + fn aspect_properties(&self) -> &[crate::AspectPropertyMetaData] { + &[] + } + } + + impl OnPremiseAspectEngine for IdleEngine { + fn data_files(&self) -> &[Arc] { + &self.data_files + } + fn refresh(&self, _identifier: Option<&str>) -> fiftyone_pipeline_core::Result<()> { + Ok(()) + } + } + + /// A data file with no update URL that polls every `interval_seconds` + /// plus up to `max_randomisation_seconds`. + fn polled_file(interval_seconds: u64, max_randomisation_seconds: u64) -> AspectEngineDataFile { + AspectEngineDataFile::new( + crate::DataFileConfiguration::memory_builder() + .polling_interval_seconds(interval_seconds) + .max_randomisation_seconds(max_randomisation_seconds) + .build(), + ) + } + + /// The delay before the first poll is drawn at random, so instances that + /// start together poll at different times. 200 draws from 600 seconds + /// give about 170 different delays. + #[test] + fn the_first_poll_is_delayed_at_random() { + let service = DataUpdateService::new(); + let data_file = polled_file(0, 600); + let delays: std::collections::BTreeSet = (0..200) + .map(|_| { + let start = Instant::now(); + service + .initial_poll_due(&data_file) + .duration_since(start) + .as_secs() + }) + .collect(); + eprintln!("{} different delays in 200", delays.len()); + assert!( + delays.len() > 100, + "{} different delays in 200: {delays:?}", + delays.len() + ); + } + + /// Each poll after the first also waits the polling interval plus a + /// random delay, as the specification says. + #[test] + fn each_later_poll_adds_a_random_delay() { + const INTERVAL_SECONDS: u64 = 1000; + let handle = ServiceHandle { + client: reqwest::blocking::Client::new(), + inner: Arc::new(Inner { + registrations: Mutex::new(Vec::new()), + wake: Condvar::new(), + shutting_down: Mutex::new(false), + }), + }; + let due = Instant::now(); + for _ in 0..5 { + let data_file = Arc::new(polled_file(INTERVAL_SECONDS, INTERVAL_SECONDS)); + let engine = Arc::new(IdleEngine { + filter: fiftyone_pipeline_core::EvidenceKeyFilterWhitelist::new( + Vec::::new(), + ), + data_files: vec![Arc::clone(&data_file)], + }); + handle + .inner + .registrations + .lock() + .unwrap() + .push(Arc::new(Registration { + engine, + data_file, + next_poll: Mutex::new(Some(due)), + watcher: Mutex::new(None), + })); + } + + handle.run_due_polls(); + + // Without a delay every next poll is due at most one interval after + // the polls ran. Five delays of under two seconds out of a thousand + // happen once in about 3 * 10^13 runs. + let undelayed = Instant::now() + Duration::from_secs(INTERVAL_SECONDS + 1); + let delayed = handle + .inner + .registrations + .lock() + .unwrap() + .iter() + .filter(|registration| { + registration + .next_poll + .lock() + .unwrap() + .is_some_and(|next| next > undelayed) + }) + .count(); + assert!(delayed > 0, "no poll after the first added a delay"); + } }