mirror of
https://github.com/clash-verge-rev/clash-verge-rev.git
synced 2026-11-08 02:52:07 +08:00
Refactor timer management and remove delay_timer dependency (#7075)
* Refactor timer management and remove delay_timer dependency
The delay_timer spawned tokio threads of cores, which not controled by expected.
* refactor: improve logging messages in Timer implementation
* fix(timer): prevent potential panic by using saturating_mul for interval calculations
* refactor timer scheduler
* chore: translate timer logs to english
* fix timer clippy warnings
* fix profile auto update refresh
* fix duplicate profile update notification
* fix: prevent potential panic in lightweight timer setup
* fix profile update listener cleanup
* Revert "fix duplicate profile update notification"
This reverts commit 4aafa94d3e.
This commit is contained in:
Tunglies
+271
-331
@@ -1,60 +1,66 @@
|
||||
use crate::{config::Config, feat, singleton, utils::resolve::is_resolve_done};
|
||||
use anyhow::{Context as _, Result};
|
||||
use crate::{config::Config, core::handle, feat, process::AsyncHandler, singleton, utils::resolve::is_resolve_done};
|
||||
use anyhow::Result;
|
||||
use clash_verge_logging::{Type, logging, logging_error};
|
||||
use delay_timer::prelude::{DelayTimer, DelayTimerBuilder, TaskBuilder};
|
||||
use parking_lot::RwLock;
|
||||
use parking_lot::{Mutex, RwLock};
|
||||
use smartstring::alias::String;
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
pin::Pin,
|
||||
sync::{
|
||||
Arc,
|
||||
atomic::{AtomicBool, AtomicU64, Ordering},
|
||||
atomic::{AtomicBool, Ordering},
|
||||
},
|
||||
time::Duration,
|
||||
};
|
||||
use tokio::time::{sleep, timeout};
|
||||
use tokio::{
|
||||
sync::mpsc,
|
||||
time::{sleep, timeout},
|
||||
};
|
||||
use tokio_stream::StreamExt as _;
|
||||
use tokio_util::time::{DelayQueue, delay_queue::Key};
|
||||
|
||||
type TaskID = u64;
|
||||
enum TimerCommand {
|
||||
Apply(HashMap<String, u64>),
|
||||
RunNow(String),
|
||||
TaskFinished(String),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TimerTask {
|
||||
pub task_id: TaskID,
|
||||
pub interval_minutes: u64,
|
||||
#[allow(unused)]
|
||||
pub last_run: i64, // Timestamp of last execution
|
||||
struct TaskState {
|
||||
key: Option<Key>,
|
||||
interval_minutes: u64,
|
||||
running: bool,
|
||||
}
|
||||
|
||||
impl TaskState {
|
||||
const fn new(key: Key, interval_minutes: u64) -> Self {
|
||||
Self {
|
||||
key: Some(key),
|
||||
interval_minutes,
|
||||
running: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct Timer {
|
||||
/// cron manager
|
||||
pub delay_timer: Arc<RwLock<DelayTimer>>,
|
||||
|
||||
/// save the current state - using RwLock for better read concurrency
|
||||
pub timer_map: Arc<RwLock<HashMap<String, TimerTask>>>,
|
||||
|
||||
/// increment id - atomic counter for better performance
|
||||
pub timer_count: AtomicU64,
|
||||
|
||||
/// Flag to mark if timer is initialized - atomic for better performance
|
||||
command_tx: mpsc::UnboundedSender<TimerCommand>,
|
||||
command_rx: Mutex<Option<mpsc::UnboundedReceiver<TimerCommand>>>,
|
||||
pub timer_map: Arc<RwLock<HashMap<String, u64>>>,
|
||||
pub initialized: AtomicBool,
|
||||
}
|
||||
|
||||
// Use singleton macro
|
||||
singleton!(Timer, TIMER_INSTANCE);
|
||||
|
||||
impl Timer {
|
||||
fn new() -> Self {
|
||||
let (command_tx, command_rx) = mpsc::unbounded_channel();
|
||||
Self {
|
||||
delay_timer: Arc::new(RwLock::new(DelayTimerBuilder::default().build())),
|
||||
command_tx,
|
||||
command_rx: Mutex::new(Some(command_rx)),
|
||||
timer_map: Arc::new(RwLock::new(HashMap::new())),
|
||||
timer_count: AtomicU64::new(1),
|
||||
initialized: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
|
||||
/// Initialize timer with better error handling and atomic operations
|
||||
pub async fn init(&self) -> Result<()> {
|
||||
// Use compare_exchange for thread-safe initialization check
|
||||
if self
|
||||
.initialized
|
||||
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
|
||||
@@ -64,76 +70,48 @@ impl Timer {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Initialize timer tasks
|
||||
let command_rx = { self.command_rx.lock().take() };
|
||||
if let Some(command_rx) = command_rx {
|
||||
let command_tx = self.command_tx.clone();
|
||||
AsyncHandler::spawn(move || async move {
|
||||
Self::run_scheduler(command_rx, command_tx).await;
|
||||
});
|
||||
}
|
||||
|
||||
if let Err(e) = self.refresh().await {
|
||||
// Reset initialization flag on error
|
||||
self.initialized.store(false, Ordering::SeqCst);
|
||||
logging_error!(Type::Timer, "Failed to initialize timer: {}", e);
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
// Log timer info first
|
||||
{
|
||||
let timer_map = self.timer_map.read();
|
||||
logging!(info, Type::Timer, "已注册的定时任务数量: {}", timer_map.len());
|
||||
|
||||
for (uid, task) in timer_map.iter() {
|
||||
logging!(debug, Type::Timer, "Registered timer task count: {}", timer_map.len());
|
||||
for (uid, interval) in timer_map.iter() {
|
||||
logging!(
|
||||
info,
|
||||
debug,
|
||||
Type::Timer,
|
||||
"注册了定时任务 - uid={}, interval={}min, task_id={}",
|
||||
"Registered timer task: uid={}, interval={}min",
|
||||
uid,
|
||||
task.interval_minutes,
|
||||
task.task_id
|
||||
interval
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let cur_timestamp = chrono::Local::now().timestamp();
|
||||
|
||||
// Collect profiles that need immediate update
|
||||
let profiles_to_update = if let Some(items) = Config::profiles().await.latest_arc().get_items() {
|
||||
items
|
||||
.iter()
|
||||
.filter_map(|item| {
|
||||
let allow_auto_update = item.option.as_ref()?.allow_auto_update.unwrap_or_default();
|
||||
if !allow_auto_update {
|
||||
return None;
|
||||
}
|
||||
|
||||
let interval = item.option.as_ref()?.update_interval? as i64;
|
||||
let updated = item.updated? as i64;
|
||||
let uid = item.uid.as_ref()?;
|
||||
|
||||
if interval > 0 && cur_timestamp - updated >= interval * 60 {
|
||||
logging!(info, Type::Timer, "需要立即更新的配置: uid={}", uid);
|
||||
Some(uid.clone())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect::<Vec<String>>()
|
||||
} else {
|
||||
Vec::new()
|
||||
};
|
||||
|
||||
// Advance tasks outside of locks to minimize lock contention
|
||||
if !profiles_to_update.is_empty() {
|
||||
logging!(
|
||||
info,
|
||||
Type::Timer,
|
||||
"需要立即更新的配置数量: {}",
|
||||
profiles_to_update.len()
|
||||
);
|
||||
let timer_map = self.timer_map.read();
|
||||
|
||||
for uid in profiles_to_update {
|
||||
if let Some(task) = timer_map.get(&uid) {
|
||||
logging!(info, Type::Timer, "立即执行任务: uid={}", uid);
|
||||
let delay_timer = self.delay_timer.write();
|
||||
if let Err(e) = delay_timer.advance_task(task.task_id) {
|
||||
logging!(warn, Type::Timer, "Failed to advance task {}: {}", uid, e);
|
||||
}
|
||||
if let Some(items) = Config::profiles().await.latest_arc().get_items() {
|
||||
for item in items.iter() {
|
||||
if let Some(option) = item.option.as_ref()
|
||||
&& let Some(allow_auto_update) = option.allow_auto_update
|
||||
&& allow_auto_update
|
||||
&& let Some(interval) = option.update_interval
|
||||
&& interval > 0
|
||||
&& let Some(uid) = item.uid.as_ref()
|
||||
&& let Some(updated) = item.updated
|
||||
&& cur_timestamp - (updated as i64) >= (interval as i64) * 60
|
||||
{
|
||||
logging!(info, Type::Timer, "Running overdue timer task immediately: uid={}", uid);
|
||||
let _ = self.command_tx.send(TimerCommand::RunNow(uid.clone()));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -142,96 +120,29 @@ impl Timer {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Refresh timer tasks with better error handling
|
||||
pub async fn refresh(&self) -> Result<()> {
|
||||
// Generate diff outside of lock to minimize lock contention
|
||||
let diff_map = self.gen_diff().await;
|
||||
let new_map = self.gen_map().await;
|
||||
|
||||
if diff_map.is_empty() {
|
||||
let mut cache = self.timer_map.write();
|
||||
if *cache == new_map {
|
||||
logging!(debug, Type::Timer, "No timer changes needed");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
logging!(info, Type::Timer, "Refreshing {} timer tasks", diff_map.len());
|
||||
logging!(
|
||||
info,
|
||||
Type::Timer,
|
||||
"Refreshing timer tasks map, count: {}",
|
||||
new_map.len()
|
||||
);
|
||||
*cache = new_map.clone();
|
||||
drop(cache);
|
||||
|
||||
// Apply changes - first collect operations to perform without holding locks
|
||||
let mut operations_to_add: Vec<(String, TaskID, u64)> = Vec::new();
|
||||
let _operations_to_remove: Vec<String> = Vec::new();
|
||||
|
||||
// Perform sync operations while holding locks
|
||||
{
|
||||
for (uid, diff) in diff_map {
|
||||
match diff {
|
||||
DiffFlag::Del(tid) => {
|
||||
self.timer_map.write().remove(&uid);
|
||||
let value = self.delay_timer.write().remove_task(tid);
|
||||
if let Err(e) = value {
|
||||
logging!(
|
||||
warn,
|
||||
Type::Timer,
|
||||
"Failed to remove task {} for uid {}: {}",
|
||||
tid,
|
||||
uid,
|
||||
e
|
||||
);
|
||||
} else {
|
||||
logging!(debug, Type::Timer, "Removed task {} for uid {}", tid, uid);
|
||||
}
|
||||
}
|
||||
DiffFlag::Add(tid, interval) => {
|
||||
let task = TimerTask {
|
||||
task_id: tid,
|
||||
interval_minutes: interval,
|
||||
last_run: chrono::Local::now().timestamp(),
|
||||
};
|
||||
|
||||
self.timer_map.write().insert(uid.clone(), task);
|
||||
operations_to_add.push((uid, tid, interval));
|
||||
}
|
||||
DiffFlag::Mod(tid, interval) => {
|
||||
// Remove old task first
|
||||
let value = self.delay_timer.write().remove_task(tid);
|
||||
if let Err(e) = value {
|
||||
logging!(
|
||||
warn,
|
||||
Type::Timer,
|
||||
"Failed to remove old task {} for uid {}: {}",
|
||||
tid,
|
||||
uid,
|
||||
e
|
||||
);
|
||||
}
|
||||
|
||||
// Then add the new one
|
||||
let task = TimerTask {
|
||||
task_id: tid,
|
||||
interval_minutes: interval,
|
||||
last_run: chrono::Local::now().timestamp(),
|
||||
};
|
||||
|
||||
self.timer_map.write().insert(uid.clone(), task);
|
||||
operations_to_add.push((uid, tid, interval));
|
||||
}
|
||||
}
|
||||
}
|
||||
} // Locks are dropped here
|
||||
|
||||
// Now perform async operations without holding locks
|
||||
let delay_timer = self.delay_timer.write();
|
||||
for (uid, tid, interval) in operations_to_add {
|
||||
if let Err(e) = self.add_task(&delay_timer, uid.clone(), tid, interval) {
|
||||
logging_error!(Type::Timer, "Failed to add task for uid {}: {}", uid, e);
|
||||
// Rollback on failure - remove from timer_map
|
||||
self.timer_map.write().remove(&uid);
|
||||
} else {
|
||||
logging!(debug, Type::Timer, "Added task {} for uid {}", tid, uid);
|
||||
}
|
||||
}
|
||||
let _ = self.command_tx.send(TimerCommand::Apply(new_map));
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Generate map of profile UIDs to update intervals
|
||||
async fn gen_map(&self) -> HashMap<String, u64> {
|
||||
let mut new_map = HashMap::new();
|
||||
|
||||
@@ -243,218 +154,254 @@ impl Timer {
|
||||
&& allow_auto_update
|
||||
&& interval > 0
|
||||
{
|
||||
logging!(
|
||||
debug,
|
||||
Type::Timer,
|
||||
"找到定时更新配置: uid={}, interval={}min",
|
||||
uid,
|
||||
interval
|
||||
);
|
||||
new_map.insert(uid.clone(), interval);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
logging!(debug, Type::Timer, "生成的定时更新配置数量: {}", new_map.len());
|
||||
new_map
|
||||
}
|
||||
|
||||
/// Generate differences between current and new timer configuration
|
||||
async fn gen_diff(&self) -> HashMap<String, DiffFlag> {
|
||||
let mut diff_map = HashMap::new();
|
||||
let new_map = self.gen_map().await;
|
||||
async fn run_scheduler(
|
||||
mut command_rx: mpsc::UnboundedReceiver<TimerCommand>,
|
||||
command_tx: mpsc::UnboundedSender<TimerCommand>,
|
||||
) {
|
||||
let mut queue = DelayQueue::new();
|
||||
let mut tasks = HashMap::new();
|
||||
|
||||
// Read lock for comparing current state
|
||||
let timer_map = self.timer_map.read();
|
||||
logging!(debug, Type::Timer, "当前 timer_map 大小: {}", timer_map.len());
|
||||
loop {
|
||||
tokio::select! {
|
||||
Some(expired) = queue.next() => {
|
||||
let uid = expired.into_inner();
|
||||
Self::run_expired_task(&mut tasks, uid, command_tx.clone());
|
||||
}
|
||||
|
||||
// Find tasks to modify or delete
|
||||
for (uid, task) in timer_map.iter() {
|
||||
match new_map.get(uid) {
|
||||
Some(&interval) if interval != task.interval_minutes => {
|
||||
// Task exists but interval changed
|
||||
logging!(
|
||||
debug,
|
||||
Type::Timer,
|
||||
"定时任务间隔变更: uid={}, 旧={}, 新={}",
|
||||
uid,
|
||||
task.interval_minutes,
|
||||
interval
|
||||
);
|
||||
diff_map.insert(uid.clone(), DiffFlag::Mod(task.task_id, interval));
|
||||
}
|
||||
None => {
|
||||
// Task no longer needed
|
||||
logging!(debug, Type::Timer, "定时任务已删除: uid={}", uid);
|
||||
diff_map.insert(uid.clone(), DiffFlag::Del(task.task_id));
|
||||
}
|
||||
_ => {
|
||||
// Task exists with same interval, no change needed
|
||||
logging!(debug, Type::Timer, "定时任务保持不变: uid={}", uid);
|
||||
command = command_rx.recv() => {
|
||||
match command {
|
||||
Some(TimerCommand::Apply(new_map)) => {
|
||||
Self::apply_timer_map(&mut queue, &mut tasks, new_map);
|
||||
}
|
||||
Some(TimerCommand::RunNow(uid)) => {
|
||||
Self::run_task_now(&mut queue, &mut tasks, uid, command_tx.clone());
|
||||
}
|
||||
Some(TimerCommand::TaskFinished(uid)) => {
|
||||
Self::finish_task(&mut queue, &mut tasks, uid);
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Find new tasks to add
|
||||
let mut next_id = self.timer_count.load(Ordering::Relaxed);
|
||||
let original_id = next_id;
|
||||
|
||||
for (uid, &interval) in new_map.iter() {
|
||||
if !timer_map.contains_key(uid) {
|
||||
logging!(
|
||||
debug,
|
||||
Type::Timer,
|
||||
"新增定时任务: uid={}, interval={}min",
|
||||
uid,
|
||||
interval
|
||||
);
|
||||
diff_map.insert(uid.clone(), DiffFlag::Add(next_id, interval));
|
||||
next_id += 1;
|
||||
}
|
||||
}
|
||||
|
||||
// Update counter only if we added new tasks
|
||||
if next_id > original_id {
|
||||
self.timer_count.store(next_id, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
logging!(debug, Type::Timer, "定时任务变更数量: {}", diff_map.len());
|
||||
diff_map
|
||||
}
|
||||
|
||||
/// Add a timer task with better error handling
|
||||
fn add_task(&self, delay_timer: &DelayTimer, uid: String, tid: TaskID, minutes: u64) -> Result<()> {
|
||||
fn apply_timer_map(
|
||||
queue: &mut DelayQueue<String>,
|
||||
tasks: &mut HashMap<String, TaskState>,
|
||||
new_map: HashMap<String, u64>,
|
||||
) {
|
||||
tasks.retain(|uid, state| {
|
||||
if new_map.contains_key(uid) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if let Some(key) = state.key.take() {
|
||||
queue.remove(&key);
|
||||
}
|
||||
logging!(debug, Type::Timer, "Removed timer task for uid={}", uid);
|
||||
false
|
||||
});
|
||||
|
||||
for (uid, interval_minutes) in new_map {
|
||||
let Some(state) = tasks.get_mut(&uid) else {
|
||||
Self::insert_task(queue, tasks, uid, interval_minutes);
|
||||
continue;
|
||||
};
|
||||
|
||||
if state.interval_minutes == interval_minutes {
|
||||
continue;
|
||||
}
|
||||
|
||||
Self::update_task_interval(queue, &uid, state, interval_minutes);
|
||||
}
|
||||
}
|
||||
|
||||
fn insert_task(
|
||||
queue: &mut DelayQueue<String>,
|
||||
tasks: &mut HashMap<String, TaskState>,
|
||||
uid: String,
|
||||
interval_minutes: u64,
|
||||
) {
|
||||
let key = Self::schedule_task(queue, &uid, interval_minutes);
|
||||
logging!(
|
||||
info,
|
||||
debug,
|
||||
Type::Timer,
|
||||
"Adding task: uid={}, id={}, interval={}min",
|
||||
"Added timer task: uid={}, interval={}min",
|
||||
uid,
|
||||
tid,
|
||||
minutes
|
||||
interval_minutes
|
||||
);
|
||||
tasks.insert(uid, TaskState::new(key, interval_minutes));
|
||||
}
|
||||
|
||||
// Create a task with reasonable retries and backoff
|
||||
let task = TaskBuilder::default()
|
||||
.set_task_id(tid)
|
||||
.set_maximum_parallel_runnable_num(1)
|
||||
.set_frequency_repeated_by_minutes(minutes)
|
||||
.spawn_async_routine(move || {
|
||||
let uid = uid.clone();
|
||||
Box::pin(async move {
|
||||
Self::wait_until_resolve_done(Duration::from_millis(5000)).await;
|
||||
Self::async_task(&uid).await;
|
||||
}) as Pin<Box<dyn std::future::Future<Output = ()> + Send>>
|
||||
})
|
||||
.context("failed to create timer task")?;
|
||||
|
||||
delay_timer.add_task(task).context("failed to add timer task")?;
|
||||
|
||||
Ok(())
|
||||
fn update_task_interval(queue: &mut DelayQueue<String>, uid: &str, state: &mut TaskState, interval_minutes: u64) {
|
||||
state.interval_minutes = interval_minutes;
|
||||
|
||||
if let Some(key) = state.key.as_ref() {
|
||||
queue.reset(key, Self::interval_duration(interval_minutes));
|
||||
} else if !state.running {
|
||||
state.key = Some(Self::schedule_task(queue, uid, interval_minutes));
|
||||
}
|
||||
|
||||
logging!(
|
||||
debug,
|
||||
Type::Timer,
|
||||
"Updated timer task interval: uid={}, interval={}min",
|
||||
uid,
|
||||
interval_minutes
|
||||
);
|
||||
}
|
||||
|
||||
fn run_expired_task(
|
||||
tasks: &mut HashMap<String, TaskState>,
|
||||
uid: String,
|
||||
command_tx: mpsc::UnboundedSender<TimerCommand>,
|
||||
) {
|
||||
let Some(state) = tasks.get_mut(&uid) else {
|
||||
return;
|
||||
};
|
||||
|
||||
state.key = None;
|
||||
if !Self::mark_task_running(state, &uid, false) {
|
||||
return;
|
||||
}
|
||||
|
||||
Self::spawn_update_task(uid, command_tx);
|
||||
}
|
||||
|
||||
fn run_task_now(
|
||||
queue: &mut DelayQueue<String>,
|
||||
tasks: &mut HashMap<String, TaskState>,
|
||||
uid: String,
|
||||
command_tx: mpsc::UnboundedSender<TimerCommand>,
|
||||
) {
|
||||
let Some(state) = tasks.get_mut(&uid) else {
|
||||
return;
|
||||
};
|
||||
|
||||
if !Self::mark_task_running(state, &uid, true) {
|
||||
return;
|
||||
}
|
||||
|
||||
if let Some(key) = state.key.take() {
|
||||
queue.remove(&key);
|
||||
}
|
||||
Self::spawn_update_task(uid, command_tx);
|
||||
}
|
||||
|
||||
fn mark_task_running(state: &mut TaskState, uid: &str, immediate: bool) -> bool {
|
||||
if !state.running {
|
||||
state.running = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
if immediate {
|
||||
logging!(
|
||||
debug,
|
||||
Type::Timer,
|
||||
"Timer task already running, skip immediate uid={}",
|
||||
uid
|
||||
);
|
||||
} else {
|
||||
logging!(debug, Type::Timer, "Timer task already running, skip uid={}", uid);
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn finish_task(queue: &mut DelayQueue<String>, tasks: &mut HashMap<String, TaskState>, uid: String) {
|
||||
let Some(state) = tasks.get_mut(&uid) else {
|
||||
return;
|
||||
};
|
||||
|
||||
state.running = false;
|
||||
let key = Self::schedule_task(queue, &uid, state.interval_minutes);
|
||||
state.key = Some(key);
|
||||
}
|
||||
|
||||
fn spawn_update_task(uid: String, command_tx: mpsc::UnboundedSender<TimerCommand>) {
|
||||
logging!(info, Type::Timer, "Starting timer task: uid={}", uid);
|
||||
AsyncHandler::spawn(move || async move {
|
||||
Self::wait_until_resolve_done(Duration::from_millis(5000)).await;
|
||||
Self::async_task(&uid).await;
|
||||
let _ = command_tx.send(TimerCommand::TaskFinished(uid));
|
||||
});
|
||||
}
|
||||
|
||||
const fn interval_duration(interval_minutes: u64) -> Duration {
|
||||
Duration::from_secs(interval_minutes.saturating_mul(60))
|
||||
}
|
||||
|
||||
fn schedule_task(queue: &mut DelayQueue<String>, uid: &str, interval_minutes: u64) -> Key {
|
||||
queue.insert(String::from(uid), Self::interval_duration(interval_minutes))
|
||||
}
|
||||
|
||||
/// Get next update time for a profile
|
||||
pub async fn get_next_update_time(&self, uid: &str) -> Option<i64> {
|
||||
logging!(info, Type::Timer, "获取下次更新时间，uid={}", uid);
|
||||
logging!(debug, Type::Timer, "Getting next update time, uid={}", uid);
|
||||
|
||||
// First extract timer task data without holding the lock across await
|
||||
let task_interval = {
|
||||
let timer_map = self.timer_map.read();
|
||||
match timer_map.get(uid) {
|
||||
Some(t) => t.interval_minutes,
|
||||
None => {
|
||||
logging!(warn, Type::Timer, "找不到对应的定时任务，uid={}", uid);
|
||||
return None;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Get the profile updated timestamp - now safe to await
|
||||
let items = {
|
||||
let profiles = Config::profiles().await;
|
||||
let profiles_guard = profiles.latest_arc();
|
||||
match profiles_guard.get_items() {
|
||||
Some(i) => i.clone(),
|
||||
None => {
|
||||
logging!(warn, Type::Timer, "获取配置列表失败");
|
||||
return None;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let profile = match items.iter().find(|item| item.uid.as_deref() == Some(uid)) {
|
||||
Some(p) => p,
|
||||
None => {
|
||||
logging!(warn, Type::Timer, "找不到对应的配置，uid={}", uid);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
let task_interval = *self.timer_map.read().get(uid)?;
|
||||
let profiles = Config::profiles().await;
|
||||
let profiles_guard = profiles.latest_arc();
|
||||
let items = profiles_guard.get_items()?;
|
||||
|
||||
let profile = items.iter().find(|item| item.uid.as_deref() == Some(uid))?;
|
||||
let updated = profile.updated.unwrap_or(0) as i64;
|
||||
|
||||
// Calculate next update time
|
||||
if updated > 0 && task_interval > 0 {
|
||||
let next_time = updated + (task_interval as i64 * 60);
|
||||
logging!(info, Type::Timer, "计算得到下次更新时间: {}, uid={}", next_time, uid);
|
||||
Some(next_time)
|
||||
if updated > 0 {
|
||||
Some(updated + (task_interval as i64 * 60))
|
||||
} else {
|
||||
logging!(
|
||||
warn,
|
||||
Type::Timer,
|
||||
"更新时间或间隔无效，updated={}, interval={}",
|
||||
updated,
|
||||
task_interval
|
||||
);
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// Emit update events for frontend notification
|
||||
fn emit_update_event(uid: &String, is_start: bool) {
|
||||
{
|
||||
if is_start {
|
||||
super::handle::Handle::notify_profile_update_started(uid);
|
||||
} else {
|
||||
super::handle::Handle::notify_profile_update_completed(uid);
|
||||
}
|
||||
if is_start {
|
||||
super::handle::Handle::notify_profile_update_started(uid);
|
||||
} else {
|
||||
super::handle::Handle::notify_profile_update_completed(uid);
|
||||
}
|
||||
}
|
||||
|
||||
/// Async task with better error handling and logging
|
||||
async fn async_task(uid: &String) {
|
||||
let task_start = std::time::Instant::now();
|
||||
logging!(info, Type::Timer, "Running timer task for profile: {}", uid);
|
||||
logging!(debug, Type::Timer, "Running timer task for profile: {}", uid);
|
||||
|
||||
match tokio::time::timeout(std::time::Duration::from_secs(40), async {
|
||||
Self::emit_update_event(uid, true);
|
||||
|
||||
let is_current = Config::profiles().await.latest_arc().current.as_ref() == Some(uid);
|
||||
logging!(info, Type::Timer, "配置 {} 是否为当前激活配置: {}", uid, is_current);
|
||||
logging!(
|
||||
debug,
|
||||
Type::Timer,
|
||||
"Profile {} is current active profile: {}",
|
||||
uid,
|
||||
is_current
|
||||
);
|
||||
|
||||
feat::update_profile(uid, None, is_current, false, false).await
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(result) => match result {
|
||||
Ok(_) => {
|
||||
let duration = task_start.elapsed().as_millis();
|
||||
logging!(
|
||||
info,
|
||||
Type::Timer,
|
||||
"Timer task completed successfully for uid: {} (took {}ms)",
|
||||
uid,
|
||||
duration
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
logging_error!(Type::Timer, "Failed to update profile uid {}: {}", uid, e);
|
||||
}
|
||||
},
|
||||
Err(_) => {
|
||||
logging_error!(Type::Timer, "Timer task timed out for uid: {}", uid);
|
||||
Ok(Ok(_)) => {
|
||||
logging!(
|
||||
info,
|
||||
Type::Timer,
|
||||
"Timer task completed for uid: {} (took {}ms)",
|
||||
uid,
|
||||
task_start.elapsed().as_millis()
|
||||
);
|
||||
handle::Handle::notify_profile_update_completed(uid);
|
||||
}
|
||||
Ok(Err(e)) => logging_error!(Type::Timer, "Failed to update profile uid {}: {}", uid, e),
|
||||
Err(_) => logging_error!(Type::Timer, "Timer task timed out for uid: {}", uid),
|
||||
}
|
||||
|
||||
// Emit completed event
|
||||
Self::emit_update_event(uid, false);
|
||||
}
|
||||
|
||||
@@ -468,10 +415,3 @@ impl Timer {
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum DiffFlag {
|
||||
Del(TaskID),
|
||||
Add(TaskID, u64),
|
||||
Mod(TaskID, u64),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user