From 232be337546e1191fb483e59862f4e6790330917 Mon Sep 17 00:00:00 2001 From: Tunglies <77394545+Tunglies@users.noreply.github.com> Date: Sat, 30 May 2026 01:54:39 +0800 Subject: [PATCH] fix(draft): serialize async updates --- crates/clash-verge-draft/Cargo.toml | 2 +- crates/clash-verge-draft/src/lib.rs | 91 ++++++--- crates/clash-verge-draft/tests/test_me.rs | 222 +++++++++++++--------- src-tauri/src/enhance/mod.rs | 2 +- 4 files changed, 203 insertions(+), 114 deletions(-) diff --git a/crates/clash-verge-draft/Cargo.toml b/crates/clash-verge-draft/Cargo.toml index 798b9b8a8..fe5ed4d56 100644 --- a/crates/clash-verge-draft/Cargo.toml +++ b/crates/clash-verge-draft/Cargo.toml @@ -11,7 +11,7 @@ harness = false [dependencies] parking_lot = { workspace = true } anyhow = { workspace = true } +tokio = { workspace = true } [dev-dependencies] criterion = { workspace = true } -tokio = { workspace = true } diff --git a/crates/clash-verge-draft/src/lib.rs b/crates/clash-verge-draft/src/lib.rs index a26f1513b..c1c5befbd 100644 --- a/crates/clash-verge-draft/src/lib.rs +++ b/crates/clash-verge-draft/src/lib.rs @@ -1,27 +1,54 @@ -use parking_lot::RwLock; +use parking_lot::Mutex; use std::sync::Arc; +use std::sync::atomic::{ + AtomicBool, + Ordering::{Acquire, Relaxed, Release}, +}; +use tokio::sync::Notify; -pub type SharedDraft = Arc; -type DraftInner = (SharedDraft, Option>); - -/// Draft 管理：committed 与 optional draft 都以 Arc> 存储， +pub type SharedDraft = Arc>; // (committed_snapshot, optional_draft_snapshot) +type DraftData = (SharedDraft, Option>); +// Retry one scheduler turn before parking on Notify for long async updates. +const DATA_MODIFY_FAST_RETRY_YIELDS: usize = 1; + #[derive(Debug)] +struct DraftInner { + data: Mutex>, + data_modifying: AtomicBool, + data_modify_notify: Notify, +} + +struct DataModifyPermit<'a>(&'a AtomicBool, &'a Notify); + +impl Drop for DataModifyPermit<'_> { + fn drop(&mut self) { + self.0.store(false, Release); + self.1.notify_one(); + } +} + +/// Draft 管理：committed 与 optional draft 都以 Arc> 存储。 +#[derive(Debug, Clone)] pub struct Draft { - inner: Arc>>, + inner: Arc>, } impl Draft { #[inline] pub fn new(data: T) -> Self { Self { - inner: Arc::new(RwLock::new((Arc::new(data), None))), + inner: Arc::new(DraftInner { + data: Mutex::new((Arc::new(Box::new(data)), None)), + data_modifying: AtomicBool::new(false), + data_modify_notify: Notify::new(), + }), } } /// 以 Arc> 的形式获取当前“已提交（正式）”数据的快照（零拷贝，仅 clone Arc） #[inline] pub fn data_arc(&self) -> SharedDraft { - let guard = self.inner.read(); + let guard = self.inner.data.lock(); Arc::clone(&guard.0) } @@ -29,7 +56,7 @@ impl Draft { /// 这也是零拷贝：只 clone Arc，不 clone T #[inline] pub fn latest_arc(&self) -> SharedDraft { - let guard = self.inner.read(); + let guard = self.inner.data.lock(); guard.1.clone().unwrap_or_else(|| Arc::clone(&guard.0)) } @@ -41,10 +68,10 @@ impl Draft { where F: FnOnce(&mut T) -> R, { - let mut guard = self.inner.write(); + let mut guard = self.inner.data.lock(); let mut draft_arc = guard.1.take().unwrap_or_else(|| Arc::clone(&guard.0)); let data_mut = Arc::make_mut(&mut draft_arc); - let result = f(data_mut); + let result = f(data_mut.as_mut()); guard.1 = Some(draft_arc); result } @@ -52,7 +79,7 @@ impl Draft { /// 将草稿提交到已提交位置（替换），并清除草稿 #[inline] pub fn apply(&self) { - let mut guard = self.inner.write(); + let mut guard = self.inner.data.lock(); if let Some(d) = guard.1.take() { guard.0 = d; } @@ -61,7 +88,7 @@ impl Draft { /// 丢弃草稿（如果存在） #[inline] pub fn discard(&self) { - let mut guard = self.inner.write(); + let mut guard = self.inner.data.lock(); guard.1 = None; } @@ -74,27 +101,47 @@ impl Draft { F: FnOnce(T) -> Fut + Send, Fut: std::future::Future> + Send, { + let _permit = self.acquire_data_modify_permit().await; let (local, original_arc) = { - let guard = self.inner.read(); + let guard = self.inner.data.lock(); let arc = Arc::clone(&guard.0); - ((*arc).clone(), arc) + ((**arc).clone(), arc) }; let (new_local, res) = f(local).await?; - let mut guard = self.inner.write(); + let mut guard = self.inner.data.lock(); if !Arc::ptr_eq(&guard.0, &original_arc) { return Err(anyhow::anyhow!( "Optimistic lock failed: Committed data has changed during async operation" )); } - guard.0 = Arc::from(new_local); + guard.0 = Arc::new(Box::new(new_local)); Ok(res) } -} -impl Clone for Draft { - fn clone(&self) -> Self { - Self { - inner: Arc::clone(&self.inner), + #[inline] + fn try_acquire_data_modify_permit(&self) -> Option> { + self.inner + .data_modifying + .compare_exchange(false, true, Acquire, Relaxed) + .ok() + .map(|_| DataModifyPermit(&self.inner.data_modifying, &self.inner.data_modify_notify)) + } + + #[inline] + async fn acquire_data_modify_permit(&self) -> DataModifyPermit<'_> { + for _ in 0..DATA_MODIFY_FAST_RETRY_YIELDS { + if let Some(permit) = self.try_acquire_data_modify_permit() { + return permit; + } + tokio::task::yield_now().await; + } + + loop { + let notified = self.inner.data_modify_notify.notified(); + if let Some(permit) = self.try_acquire_data_modify_permit() { + return permit; + } + notified.await; } } } diff --git a/crates/clash-verge-draft/tests/test_me.rs b/crates/clash-verge-draft/tests/test_me.rs index d9708c627..c7efa4eef 100644 --- a/crates/clash-verge-draft/tests/test_me.rs +++ b/crates/clash-verge-draft/tests/test_me.rs @@ -3,8 +3,10 @@ mod tests { use anyhow::anyhow; use clash_verge_draft::Draft; use std::future::Future; - use std::pin::Pin; - use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker}; + use std::sync::Arc; + use std::time::Duration; + use tokio::sync::Notify; + use tokio::time::sleep; #[derive(Clone, Debug, Default, PartialEq)] struct IVerge { @@ -12,69 +14,48 @@ mod tests { enable_tun_mode: Option, } - // Minimal single-threaded executor for immediately-ready futures - fn block_on_ready(fut: F) -> F::Output { - fn no_op_raw_waker() -> RawWaker { - fn clone(_: *const ()) -> RawWaker { - no_op_raw_waker() - } - fn wake(_: *const ()) {} - fn wake_by_ref(_: *const ()) {} - fn drop(_: *const ()) {} - static VTABLE: RawWakerVTable = RawWakerVTable::new(clone, wake, wake_by_ref, drop); - RawWaker::new(std::ptr::null(), &VTABLE) - } + const STRESS_LEVELS: [usize; 6] = [8, 16, 32, 64, 128, 256]; - let waker = unsafe { Waker::from_raw(no_op_raw_waker()) }; - let mut cx = Context::from_waker(&waker); - let mut fut = Box::pin(fut); - loop { - match Pin::as_mut(&mut fut).poll(&mut cx) { - Poll::Ready(v) => return v, - Poll::Pending => std::thread::yield_now(), - } + fn verge(enable_auto_launch: bool, enable_tun_mode: bool) -> IVerge { + IVerge { + enable_auto_launch: Some(enable_auto_launch), + enable_tun_mode: Some(enable_tun_mode), } } + fn block_on_ready(fut: F) -> F::Output { + tokio::runtime::Runtime::new().unwrap().block_on(fut) + } + #[test] fn test_draft_basic_flow() { - let verge = IVerge { - enable_auto_launch: Some(true), - enable_tun_mode: Some(false), - }; - let draft = Draft::new(verge); + let draft = Draft::new(verge(true, false)); - // 读取正式数据（data_arc） { let data = draft.data_arc(); assert_eq!(data.enable_auto_launch, Some(true)); assert_eq!(data.enable_tun_mode, Some(false)); } - // 修改草稿（使用 edit_draft） draft.edit_draft(|d| { d.enable_auto_launch = Some(false); d.enable_tun_mode = Some(true); }); - // 正式数据未变 { let data = draft.data_arc(); assert_eq!(data.enable_auto_launch, Some(true)); assert_eq!(data.enable_tun_mode, Some(false)); } - // 草稿已变 { let latest = draft.latest_arc(); assert_eq!(latest.enable_auto_launch, Some(false)); assert_eq!(latest.enable_tun_mode, Some(true)); } - // 提交草稿 draft.apply(); - // 正式数据已更新 { let data = draft.data_arc(); assert_eq!(data.enable_auto_launch, Some(false)); @@ -91,18 +72,13 @@ mod tests { assert_eq!(latest.enable_tun_mode, Some(true)); } - // 丢弃草稿 draft.discard(); - // 丢弃后再次创建草稿，会从已提交重新 clone { draft.edit_draft(|d| { - // 原 committed 是 enable_auto_launch = Some(false) assert_eq!(d.enable_auto_launch, Some(false)); - // 再修改一下 d.enable_tun_mode = Some(false); }); - // 草稿中值已修改，但正式数据仍是 apply 后的值 let data = draft.data_arc(); assert_eq!(data.enable_auto_launch, Some(false)); assert_eq!(data.enable_tun_mode, Some(true)); @@ -111,41 +87,29 @@ mod tests { #[test] fn test_arc_pointer_behavior_on_edit_and_apply() { - let draft = Draft::new(IVerge { - enable_auto_launch: Some(true), - enable_tun_mode: Some(false), - }); + let draft = Draft::new(verge(true, false)); - // 初始 latest == committed let committed = draft.data_arc(); let latest = draft.latest_arc(); - assert!(std::sync::Arc::ptr_eq(&committed, &latest)); + assert!(Arc::ptr_eq(&committed, &latest)); - // 第一次 edit：由于与 committed 共享，Arc::make_mut 会克隆 draft.edit_draft(|d| d.enable_tun_mode = Some(true)); let committed_after_first_edit = draft.data_arc(); let draft_after_first_edit = draft.latest_arc(); - assert!(!std::sync::Arc::ptr_eq( - &committed_after_first_edit, - &draft_after_first_edit - )); - // 提交会把 committed 指向草稿的 Arc - let prev_draft_ptr = std::sync::Arc::as_ptr(&draft_after_first_edit); + assert!(!Arc::ptr_eq(&committed_after_first_edit, &draft_after_first_edit)); + let prev_draft_ptr = Arc::as_ptr(&draft_after_first_edit); draft.apply(); let committed_after_apply = draft.data_arc(); - assert_eq!(std::sync::Arc::as_ptr(&committed_after_apply), prev_draft_ptr); + assert_eq!(Arc::as_ptr(&committed_after_apply), prev_draft_ptr); - // 第二次编辑：此时草稿唯一持有（无其它引用），不应再克隆 - // 获取草稿 Arc 的指针并立即丢弃本地引用，避免增加 strong_count draft.edit_draft(|d| d.enable_auto_launch = Some(false)); let latest1 = draft.latest_arc(); - let latest1_ptr = std::sync::Arc::as_ptr(&latest1); - drop(latest1); // 确保只有 Draft 内部持有草稿 Arc + let latest1_ptr = Arc::as_ptr(&latest1); + drop(latest1); - // 再次编辑（unique，Arc::make_mut 不应克隆） draft.edit_draft(|d| d.enable_tun_mode = Some(false)); let latest2 = draft.latest_arc(); - let latest2_ptr = std::sync::Arc::as_ptr(&latest2); + let latest2_ptr = Arc::as_ptr(&latest2); assert_eq!(latest1_ptr, latest2_ptr, "Unique edit should not clone Arc"); assert_eq!(latest2.enable_auto_launch, Some(false)); @@ -154,22 +118,17 @@ mod tests { #[test] fn test_discard_restores_latest_to_committed() { - let draft = Draft::new(IVerge { - enable_auto_launch: Some(false), - enable_tun_mode: Some(false), - }); + let draft = Draft::new(verge(false, false)); - // 创建草稿并修改 draft.edit_draft(|d| d.enable_auto_launch = Some(true)); let committed = draft.data_arc(); let latest = draft.latest_arc(); - assert!(!std::sync::Arc::ptr_eq(&committed, &latest)); + assert!(!Arc::ptr_eq(&committed, &latest)); - // 丢弃草稿后 latest 应回到 committed draft.discard(); let committed2 = draft.data_arc(); let latest2 = draft.latest_arc(); - assert!(std::sync::Arc::ptr_eq(&committed2, &latest2)); + assert!(Arc::ptr_eq(&committed2, &latest2)); assert_eq!(latest2.enable_auto_launch, Some(false)); } @@ -187,23 +146,13 @@ mod tests { #[test] fn test_with_data_modify_ok_and_replaces_committed() { - let draft = Draft::new(IVerge { - enable_auto_launch: Some(false), - enable_tun_mode: Some(false), - }); + let draft = Draft::new(verge(false, false)); - // 使用 with_data_modify 异步（立即就绪）地更新 committed let res = block_on_ready(draft.with_data_modify(|mut v| async move { v.enable_auto_launch = Some(true); Ok((v, "done")) })); - assert_eq!( - { - #[allow(clippy::unwrap_used)] - res.unwrap() - }, - "done" - ); + assert_eq!(res.unwrap(), "done"); let committed = draft.data_arc(); assert_eq!(committed.enable_auto_launch, Some(true)); @@ -211,7 +160,20 @@ mod tests { } #[test] - fn test_with_data_modify_error_propagation() { + fn test_with_data_modify_keeps_boxed_data_shape() { + let draft = Draft::new(Box::new(verge(false, false))); + + block_on_ready(draft.with_data_modify(|mut v| async move { + v.enable_auto_launch = Some(true); + Ok((v, ())) + })) + .unwrap(); + + assert_eq!(draft.data_arc().enable_auto_launch, Some(true)); + } + + #[test] + fn test_with_data_modify_error_releases_permit() { let draft = Draft::new(IVerge::default()); #[allow(clippy::unwrap_used)] @@ -219,42 +181,122 @@ mod tests { .unwrap_err(); assert_eq!(format!("{err}"), "boom"); + + #[allow(clippy::unwrap_used)] + block_on_ready(draft.with_data_modify(|mut v| async move { + v.enable_auto_launch = Some(true); + Ok((v, ())) + })) + .unwrap(); + + assert_eq!(draft.data_arc().enable_auto_launch, Some(true)); + } + + #[test] + fn test_with_data_modify_serialized_under_stress_levels() { + for task_count in STRESS_LEVELS { + run_with_data_modify_stress(task_count, Duration::from_millis(1)); + } + } + + #[test] + fn test_with_data_modify_keeps_apply_conflict_detection() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let draft = Arc::new(Draft::new(verge(false, false))); + let started = Arc::new(Notify::new()); + let finish = Arc::new(Notify::new()); + + rt.block_on(async { + let task = { + let draft = Arc::clone(&draft); + let started = Arc::clone(&started); + let finish = Arc::clone(&finish); + tokio::spawn(async move { + draft + .with_data_modify(|mut v| async move { + started.notify_one(); + finish.notified().await; + v.enable_auto_launch = Some(true); + Ok((v, ())) + }) + .await + }) + }; + + started.notified().await; + draft.edit_draft(|d| d.enable_tun_mode = Some(true)); + draft.apply(); + finish.notify_one(); + + let err = task.await.expect("task panicked").unwrap_err(); + assert_eq!( + format!("{err}"), + "Optimistic lock failed: Committed data has changed during async operation" + ); + }); + + let committed = draft.data_arc(); + assert_eq!(committed.enable_auto_launch, Some(false)); + assert_eq!(committed.enable_tun_mode, Some(true)); + } + + fn run_with_data_modify_stress(task_count: usize, delay: Duration) { + let rt = tokio::runtime::Runtime::new().unwrap(); + let draft = Arc::new(Draft::new(verge(false, false))); + + rt.block_on(async { + let mut handles = Vec::with_capacity(task_count); + for _ in 0..task_count { + let draft = Arc::clone(&draft); + handles.push(tokio::spawn(async move { + draft + .with_data_modify(|mut v| async move { + if !delay.is_zero() { + sleep(delay).await; + } + v.enable_auto_launch = Some(!v.enable_auto_launch.unwrap_or(false)); + Ok((v, ())) + }) + .await + })); + } + + for handle in handles { + handle.await.expect("task panicked").expect("with_data_modify failed"); + } + + Ok::<(), anyhow::Error>(()) + }) + .unwrap(); } #[test] fn test_with_data_modify_does_not_touch_existing_draft() { - let draft = Draft::new(IVerge { - enable_auto_launch: Some(false), - enable_tun_mode: Some(false), - }); + let draft = Draft::new(verge(false, false)); - // 创建草稿并修改 draft.edit_draft(|d| { d.enable_auto_launch = Some(true); d.enable_tun_mode = Some(true); }); let draft_before = draft.latest_arc(); - let draft_before_ptr = std::sync::Arc::as_ptr(&draft_before); + let draft_before_ptr = Arc::as_ptr(&draft_before); - // 同时通过 with_data_modify 修改 committed #[allow(clippy::unwrap_used)] block_on_ready(draft.with_data_modify(|mut v| async move { - v.enable_auto_launch = Some(false); // 与草稿不同 + v.enable_auto_launch = Some(false); Ok((v, ())) })) .unwrap(); - // 草稿应保持不变 let draft_after = draft.latest_arc(); assert_eq!( - std::sync::Arc::as_ptr(&draft_after), + Arc::as_ptr(&draft_after), draft_before_ptr, "Existing draft should not be replaced by with_data_modify" ); assert_eq!(draft_after.enable_auto_launch, Some(true)); assert_eq!(draft_after.enable_tun_mode, Some(true)); - // 丢弃草稿后 latest == committed，且 committed 为异步修改结果 draft.discard(); let latest = draft.latest_arc(); assert_eq!(latest.enable_auto_launch, Some(false)); diff --git a/src-tauri/src/enhance/mod.rs b/src-tauri/src/enhance/mod.rs index 23f0efb05..b38a0fd52 100644 --- a/src-tauri/src/enhance/mod.rs +++ b/src-tauri/src/enhance/mod.rs @@ -106,7 +106,7 @@ async fn get_config_values() -> ConfigValues { ref verge_http_enabled, ref enable_dns_settings, .. - } = *verge_arc; + } = **verge_arc; let (clash_core, enable_tun, enable_builtin, socks_enabled, http_enabled, enable_dns_settings) = ( Some(verge_arc.get_valid_clash_core()),