Add a config option to distribute intervals equidistantly

Author: SpriteOvO

src/config/mod.rs

@@ -24,6 +24,10 @@ use crate::{
 pub struct Config {
     #[serde(with = "humantime_serde")]
     pub interval: Duration,
+    // Distribute intervals equidistantly, which helps avoid periodic CPU peaks and request peaks
+    // when there are many tasks.
+    #[serde(default)]
+    pub equidistant_intervals: bool,
     reporter: Accessor<Option<ConfigReporterRaw>>,
     #[serde(default)]
     platform: Accessor<PlatformGlobal>,
@@ -102,7 +106,7 @@ impl Config {
                     name.clone(),
                     SubscriptionRef {
                         platform: &subscription.platform,
-                        interval: subscription.interval,
+                        interval: subscription.interval.unwrap_or(self.interval),
                         notify: subscription
                             .notify_ref
                             .iter()
@@ -184,7 +188,7 @@ pub struct SubscriptionRaw {
 #[derive(Debug, PartialEq)]
 pub struct SubscriptionRef<'a> {
     pub platform: &'a Accessor<SourceConfig>,
-    pub interval: Option<Duration>,
+    pub interval: Duration,
     pub notify: Vec<Accessor<NotifierConfig>>,
 }
 
@@ -378,6 +382,7 @@ mod tests {
         Config::parse_for_test(
             r#"
 interval = '1min'
+equidistant_intervals = true
 reporter = { log = { notify = ["meow"] }, heartbeat = { type = "HttpGet", url = "https://example.com/", interval = '1min' } } 
 
 [platform.QQ.account.MyQQ]
@@ -412,6 +417,7 @@ notify = ["meow", "woof", { ref = "woof", id = 123 }]
             |c| {
                 assert_eq!(c.unwrap(), &Config {
                     interval: Duration::from_secs(60), // 1min
+                    equidistant_intervals: true,
                     reporter: Accessor::new(Some(ConfigReporterRaw {
                         log: Accessor::new(Some(ConfigReporterLog {
                             notify_ref: vec![NotifyRef::Direct("meow".into())],
@@ -668,7 +674,7 @@ notify = ["meow", { ref = "woof", thread_id = 114 }, { ref = "woof", switch = {
                             platform: &Accessor::new(SourceConfig::BilibiliLive(Accessor::new(
                                 bilibili::source::live::ConfigParams { user_id: 123456 }
                             ))),
-                            interval: None,
+                            interval: Duration::from_secs(60),
                             notify: vec![
                                 Accessor::new(NotifierConfig::Telegram(Accessor::new(
                                     telegram::notify::ConfigParams {

src/lib.rs

@@ -33,14 +33,26 @@ pub async fn run(args: cli::Args) -> anyhow::Result<()> {
     )
     .await?;
 
-    let subscription_tasks = config.subscriptions().map(|(name, subscription)| {
-        Box::new(TaskSubscription::new(
-            name,
-            subscription.interval.unwrap_or(config.interval),
-            subscription.notify,
-            subscription.platform,
-        )) as Box<dyn Task>
+    let initial_offsets = config.equidistant_intervals.then(|| {
+        task::equidistant_intervals(
+            config
+                .subscriptions()
+                .map(|(_, subscription)| subscription.interval),
+        )
     });
+
+    let subscription_tasks = config
+        .subscriptions()
+        .enumerate()
+        .map(|(i, (name, subscription))| {
+            Box::new(TaskSubscription::new(
+                name,
+                subscription.interval,
+                initial_offsets.as_ref().map(|v| *v.get(i).unwrap()),
+                subscription.notify,
+                subscription.platform,
+            )) as Box<dyn Task>
+        });
     let reporter_task = config
         .reporter()
         .map(|params| Box::new(TaskReporter::new(params)) as Box<dyn Task>);

src/task/equidistant.rs

@@ -0,0 +1,55 @@
+use std::{collections::HashMap, time::Duration};
+
+pub fn equidistant_intervals(intervals: impl IntoIterator<Item = Duration>) -> Vec<Duration> {
+    let intervals = intervals.into_iter().collect::<Vec<_>>();
+    let mut states = intervals
+        .iter()
+        .fold(HashMap::new(), |mut acc, dur| {
+            *acc.entry(dur).or_insert(0) += 1;
+            acc
+        })
+        .into_iter()
+        .map(|(dur, num)| (*dur, (dur.div_f32(num as f32), 0)))
+        .collect::<HashMap<_, _>>();
+    intervals
+        .into_iter()
+        .map(|dur| {
+            let (offset, occurrences) = states.get_mut(&dur).unwrap();
+            let specific_offset = *offset * (*occurrences);
+            *occurrences += 1;
+            specific_offset
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use std::iter::repeat_n;
+
+    use super::*;
+
+    #[test]
+    fn equidistant_intervals_algorithm() {
+        let sec = Duration::from_secs;
+        let min = |mins: u64| sec(mins * 60);
+
+        assert_eq!(
+            equidistant_intervals(repeat_n(min(1), 10).chain(repeat_n(min(10), 3))),
+            [
+                Duration::ZERO,
+                sec(6) * 1,
+                sec(6) * 2,
+                sec(6) * 3,
+                sec(6) * 4,
+                sec(6) * 5,
+                sec(6) * 6,
+                sec(6) * 7,
+                sec(6) * 8,
+                sec(6) * 9,
+                Duration::ZERO,
+                sec(200) * 1,
+                sec(200) * 2,
+            ]
+        );
+    }
+}

src/task/mod.rs

@@ -1,8 +1,10 @@
+mod equidistant;
 mod reporter;
 mod subscription;
 
 use std::{future::Future, pin::Pin};
 
+pub use equidistant::equidistant_intervals;
 pub use reporter::TaskReporter;
 use spdlog::prelude::*;
 pub use subscription::TaskSubscription;

src/task/subscription.rs

@@ -1,5 +1,6 @@
 use std::{fmt::Display, future::Future, pin::Pin, time::Duration};
 
+use humantime_serde::re::humantime;
 use spdlog::prelude::*;
 use tokio::{sync::mpsc, time::MissedTickBehavior};
 
@@ -13,6 +14,7 @@ use crate::{
 pub struct TaskSubscription {
     name: String,
     interval: Duration,
+    initial_offset: Option<Duration>,
     notifiers: Vec<Box<dyn NotifierTrait>>,
     sourcer: Option<Sourcer>, // took when the task is running
 }
@@ -21,52 +23,74 @@ impl TaskSubscription {
     pub fn new(
         name: String,
         interval: Duration,
+        initial_offset: Option<Duration>,
         notify: Vec<Accessor<NotifierConfig>>,
         source_platform: &Accessor<SourceConfig>,
     ) -> Self {
+        trace!(
+            "task subscription '{name}' created, source '{source_platform}', interval {} (initial offset {})",
+            humantime::format_duration(interval),
+            humantime::format_duration(initial_offset.unwrap_or_default())
+        );
         Self {
             name,
             interval,
+            initial_offset,
             notifiers: notify.into_iter().map(notifier).collect(),
             sourcer: Some(sourcer(source_platform)),
         }
     }
 
     // Handler for poll-based subscription
     async fn continuous_fetch(&mut self, fetcher: Box<dyn FetcherTrait>) {
-        let mut interval = tokio::time::interval(self.interval);
-        interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
-        interval.reset_immediately();
-
         let mut last_status = Status::empty();
 
-        loop {
-            interval.tick().await;
-
-            let Ok(mut status) = fetcher.fetch_status().await.inspect_err(|err| {
-                error!(
-                    "failed to fetch status for '{}' on '{}': {err}",
-                    self.name, fetcher
-                )
-            }) else {
-                continue;
-            };
-
-            status.sort();
+        // Fetch for the first time immediately
+        self.continuous_fetch_once(&*fetcher, &mut last_status)
+            .await;
 
-            trace!(
-                "status of '{}' on '{fetcher}' now is '{status:?}'",
-                self.name
-            );
+        if let Some(initial_offset) = self.initial_offset {
+            tokio::time::sleep(initial_offset).await;
+        }
 
-            let notifications = status.generate_notifications(&last_status);
-            self.notify(notifications, &fetcher).await;
+        let mut interval = tokio::time::interval(self.interval);
+        interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
 
-            last_status.update_incrementally(status);
-            trace!("subscription '{}' updated once", self.name);
+        loop {
+            interval.tick().await;
+            self.continuous_fetch_once(&*fetcher, &mut last_status)
+                .await;
         }
     }
 
+    async fn continuous_fetch_once(
+        &mut self,
+        fetcher: &dyn FetcherTrait,
+        last_status: &mut Status,
+    ) {
+        let Ok(mut status) = fetcher.fetch_status().await.inspect_err(|err| {
+            error!(
+                "failed to fetch status for '{}' on '{}': {err}",
+                self.name, fetcher
+            )
+        }) else {
+            return;
+        };
+
+        status.sort();
+
+        trace!(
+            "status of '{}' on '{fetcher}' now is '{status:?}'",
+            self.name
+        );
+
+        let notifications = status.generate_notifications(last_status);
+        self.notify(notifications, &fetcher).await;
+
+        last_status.update_incrementally(status);
+        trace!("subscription '{}' updated once", self.name);
+    }
+
     // Handler for listen-based subscription
     async fn continuous_wait(
         &mut self,