fix(draft): serialize async updates

This commit is contained in:
Tunglies
2026-05-30 13:31:10 +08:00
parent 3444d594ac
commit 232be33754
4 changed files with 203 additions and 114 deletions
+1 -1
View File
@@ -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 }
+69 -22
View File
@@ -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<T> = Arc<T>;
type DraftInner<T> = (SharedDraft<T>, Option<SharedDraft<T>>);
/// Draft 管理:committed 与 optional draft 都以 Arc<Box<T>> 存储,
pub type SharedDraft<T> = Arc<Box<T>>;
// (committed_snapshot, optional_draft_snapshot)
type DraftData<T> = (SharedDraft<T>, Option<SharedDraft<T>>);
// Retry one scheduler turn before parking on Notify for long async updates.
const DATA_MODIFY_FAST_RETRY_YIELDS: usize = 1;
#[derive(Debug)]
struct DraftInner<T> {
data: Mutex<DraftData<T>>,
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<Box<T>> 存储。
#[derive(Debug, Clone)]
pub struct Draft<T> {
inner: Arc<RwLock<DraftInner<T>>>,
inner: Arc<DraftInner<T>>,
}
impl<T: Clone> Draft<T> {
#[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<Box<T>> 的形式获取当前“已提交(正式)”数据的快照(零拷贝,仅 clone Arc)
#[inline]
pub fn data_arc(&self) -> SharedDraft<T> {
let guard = self.inner.read();
let guard = self.inner.data.lock();
Arc::clone(&guard.0)
}
@@ -29,7 +56,7 @@ impl<T: Clone> Draft<T> {
/// 这也是零拷贝:只 clone Arc,不 clone T
#[inline]
pub fn latest_arc(&self) -> SharedDraft<T> {
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<T: Clone> Draft<T> {
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<T: Clone> Draft<T> {
/// 将草稿提交到已提交位置(替换),并清除草稿
#[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<T: Clone> Draft<T> {
/// 丢弃草稿(如果存在)
#[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<T: Clone> Draft<T> {
F: FnOnce(T) -> Fut + Send,
Fut: std::future::Future<Output = Result<(T, R), anyhow::Error>> + 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<T: Clone> Clone for Draft<T> {
fn clone(&self) -> Self {
Self {
inner: Arc::clone(&self.inner),
#[inline]
fn try_acquire_data_modify_permit(&self) -> Option<DataModifyPermit<'_>> {
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;
}
}
}
+132 -90
View File
@@ -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<bool>,
}
// Minimal single-threaded executor for immediately-ready futures
fn block_on_ready<F: Future>(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<F: Future>(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));
+1 -1
View File
@@ -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()),