You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/10/26 07:40:46 UTC
(opendal-yinyang) 02/02: fix(mount): broadcast open gate completion and atomic gate registration
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/linux-fuse-writable
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
commit ba8cd4143ccf0b039d0726a69f96e95d4d9266a9
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Oct 26 15:39:50 2026 +0800
fix(mount): broadcast open gate completion and atomic gate registration
A single semaphore permit wakes only one waiter, and a gap between node lookup and gate registration allows a second materialization. Keep both in one nodes->opening critical section and close the gate as a broadcast: every waiter retries by rechecking the node map and only one becomes the next owner. Deterministic tests cover three concurrent opens under success, first-open failure, and cancelled first open.
Co-authored-by: Chojan Shang <ps...@apache.org>
---
specs/mount-runtime.md | 10 +++-
src/mount.rs | 103 ++++++++++++++++++--------------
tests/mount.rs | 157 ++++++++++++++++++++++++++++++++++---------------
3 files changed, 176 insertions(+), 94 deletions(-)
diff --git a/specs/mount-runtime.md b/specs/mount-runtime.md
index 105e8fb..95622d8 100644
--- a/specs/mount-runtime.md
+++ b/specs/mount-runtime.md
@@ -13,9 +13,13 @@ truncate, append and fsync are serialized per node; append chooses its offset
under the same lock. Different nodes do not share an I/O lock. Initial
materialization is serialized per NodeId by a short-lived open gate, and the
instance node map is not held during a remote observation or staging read.
-Different nodes can therefore complete their first opens independently. Refresh
-and reclaim use the map for lookup and reference accounting; data-plane calls
-keep per-node locks. Cancelling a first open releases its gate; an incomplete
+The node lookup and gate registration share one critical section, so a
+completion cannot slip between them and start a second materialization.
+Success, failure and cancellation close the gate to wake every waiter for a
+retry; a retry rechecks the node map and either joins the materialized node or
+becomes the next single owner. Different nodes can therefore complete their
+first opens independently. Refresh and reclaim use the map for lookup and
+reference accounting; data-plane calls keep per-node locks. An incomplete
local open is never exposed and is discarded when staging is reopened. Both
read-only and writable references currently use full-file staging.
diff --git a/src/mount.rs b/src/mount.rs
index 1aabd54..125e448 100644
--- a/src/mount.rs
+++ b/src/mount.rs
@@ -38,7 +38,8 @@ impl Drop for OpenGate {
if let Ok(mut opening) = self.inner.opening.lock() {
opening.remove(&self.id);
}
- self.ready.add_permits(1);
+ // Close wakes every waiter; a single permit would strand later waiters.
+ self.ready.close();
}
}
struct NodeGuard {
@@ -58,7 +59,7 @@ impl Drop for NodeGuard {
struct Inner {
runtime: Runtime,
read_only: bool,
- nodes: Mutex<BTreeMap<NodeId, Arc<Node>>>,
+ nodes: StdMutex<BTreeMap<NodeId, Arc<Node>>>,
opening: StdMutex<BTreeMap<NodeId, Arc<Semaphore>>>,
}
/// A mount instance owns one recoverable staged version per node, not per open.
@@ -103,7 +104,7 @@ impl Mount {
inner: Arc::new(Inner {
runtime,
read_only,
- nodes: Mutex::new(nodes),
+ nodes: StdMutex::new(nodes),
opening: StdMutex::new(BTreeMap::new()),
}),
})
@@ -118,7 +119,7 @@ impl Mount {
/// `None` means this mount has no shared handle for the node.
pub async fn node_status(&self, id: NodeId) -> Result<Option<HandleStatus>> {
let guard = {
- let nodes = self.inner.nodes.lock().await;
+ let nodes = self.inner.nodes.lock().map_err(Error::local)?;
nodes.get(&id).cloned().map(NodeGuard::new)
};
let Some(guard) = guard else {
@@ -136,34 +137,32 @@ impl Mount {
return Err(Error::State(ErrorKind::ReadOnly, "mount is read-only"));
}
loop {
- let existing = {
- let nodes = self.inner.nodes.lock().await;
- nodes.get(&id).map(|node| {
+ // Cache lookup and gate registration must be one critical section.
+ // Always take nodes then opening, and never hold either across await.
+ let (existing, gate) = {
+ let nodes = self.inner.nodes.lock().map_err(Error::local)?;
+ if let Some(node) = nodes.get(&id) {
node.references.fetch_add(1, Ordering::Relaxed);
- node.clone()
- })
+ (Some(node.clone()), None)
+ } else {
+ let mut opening = self.inner.opening.lock().map_err(Error::local)?;
+ match opening.get(&id) {
+ Some(ready) => (None, Some((ready.clone(), true))),
+ None => {
+ let ready = Arc::new(Semaphore::new(0));
+ opening.insert(id, ready.clone());
+ (None, Some((ready, false)))
+ }
+ }
+ }
};
if let Some(node) = existing {
return Ok(MountFile { node, writable });
}
-
- let (waiting, ready) = {
- let mut opening = self.inner.opening.lock().map_err(Error::local)?;
- match opening.get(&id) {
- Some(ready) => (true, ready.clone()),
- None => {
- let ready = Arc::new(Semaphore::new(0));
- opening.insert(id, ready.clone());
- (false, ready)
- }
- }
- };
+ let (ready, waiting) = gate.expect("a waiter or owner always carries the node gate");
if waiting {
- ready
- .acquire()
- .await
- .map_err(|_| Error::local("mount open coordination closed"))?
- .forget();
+ // Closing the gate is a broadcast completion signal, not an error.
+ let _ = ready.acquire().await;
continue;
}
@@ -184,7 +183,7 @@ impl Mount {
references: AtomicUsize::new(0),
});
{
- let mut nodes = self.inner.nodes.lock().await;
+ let mut nodes = self.inner.nodes.lock().map_err(Error::local)?;
nodes.insert(id, node.clone());
node.references.fetch_add(1, Ordering::Relaxed);
}
@@ -197,7 +196,7 @@ impl Mount {
/// OS cache invalidation outside this call; no OS callbacks run under locks.
pub async fn refresh(&self, id: NodeId) -> Result<()> {
let guard = {
- let nodes = self.inner.nodes.lock().await;
+ let nodes = self.inner.nodes.lock().map_err(Error::local)?;
nodes.get(&id).cloned().map(NodeGuard::new)
};
let Some(guard) = guard else {
@@ -225,7 +224,7 @@ impl Mount {
/// so independent nodes can finish; the original failures remain in status.
pub async fn fsync(&self) -> Result<()> {
let nodes: Vec<_> = {
- let nodes = self.inner.nodes.lock().await;
+ let nodes = self.inner.nodes.lock().map_err(Error::local)?;
nodes.values().cloned().map(NodeGuard::new).collect()
};
let mut failure = None;
@@ -239,28 +238,42 @@ impl Mount {
/// Forget only an unreferenced, clean node. Failed and pending nodes remain
/// recoverable even after the OS releases all references.
pub async fn reclaim(&self, id: NodeId) -> Result<()> {
- let mut nodes = self.inner.nodes.lock().await;
- let Some(node) = nodes.get(&id) else {
- return Ok(());
+ let node = {
+ let nodes = self.inner.nodes.lock().map_err(Error::local)?;
+ let Some(node) = nodes.get(&id) else {
+ return Ok(());
+ };
+ if node.references.load(Ordering::Relaxed) != 0 {
+ return Err(Error::State(
+ ErrorKind::Busy,
+ "node still has open references",
+ ));
+ }
+ node.clone()
};
- if node.references.load(Ordering::Relaxed) != 0 {
- return Err(Error::State(
- ErrorKind::Busy,
- "node still has open references",
- ));
+ {
+ let mut handle = node.handle.lock().await;
+ let state = handle.status().await?;
+ if state.pending || state.frozen || state.error.is_some() {
+ return Err(Error::State(
+ ErrorKind::Busy,
+ "node has retained writes or failures",
+ ));
+ }
}
- let handle = node.handle.lock().await;
- let state = handle.status().await?;
- if state.pending || state.frozen || state.error.is_some() {
+ let mut nodes = self.inner.nodes.lock().map_err(Error::local)?;
+ let current = nodes.get(&id).cloned();
+ let Some(current) = current else {
+ return Ok(());
+ };
+ if current.references.load(Ordering::Relaxed) != 0 || !Arc::ptr_eq(¤t, &node) {
return Err(Error::State(
ErrorKind::Busy,
- "node has retained writes or failures",
+ "node state changed during reclaim",
));
}
- drop(handle);
- let node = nodes
- .remove(&id)
- .expect("node exists under the namespace lock");
+ let _ = nodes.remove(&id);
+ drop(nodes);
node.handle.lock().await.close().await
}
}
diff --git a/tests/mount.rs b/tests/mount.rs
index 7b05420..aeb1c26 100644
--- a/tests/mount.rs
+++ b/tests/mount.rs
@@ -36,6 +36,7 @@ struct PausingObserveAuthority {
entered: Semaphore,
release: Semaphore,
pause_next_observe: AtomicBool,
+ fail_next_observe: AtomicBool,
}
impl PausingObserveAuthority {
fn new(inner: Arc<dyn Authority>) -> Self {
@@ -44,11 +45,15 @@ impl PausingObserveAuthority {
entered: Semaphore::new(0),
release: Semaphore::new(0),
pause_next_observe: AtomicBool::new(false),
+ fail_next_observe: AtomicBool::new(false),
}
}
fn pause_next_observe(&self) {
self.pause_next_observe.store(true, Ordering::SeqCst);
}
+ fn fail_next_observe(&self) {
+ self.fail_next_observe.store(true, Ordering::SeqCst);
+ }
}
impl std::fmt::Debug for PausingObserveAuthority {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -74,6 +79,13 @@ impl Authority for PausingObserveAuthority {
self.entered.add_permits(1);
self.release.acquire().await.unwrap().forget();
}
+ if self.fail_next_observe.swap(false, Ordering::SeqCst) {
+ return Err(yinyang::core::Error::new(
+ yinyang::core::ErrorKind::Storage,
+ "test observe",
+ "injected first-open failure",
+ ));
+ }
self.inner.observe_latest().await
})
}
@@ -306,8 +318,14 @@ async fn flush_publishes_but_reference_release_does_not() {
assert!(mount.node_status(pending).await.unwrap().is_none());
}
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn independent_node_opens_do_not_share_the_namespace_lock() {
+async fn pausing_mount() -> (
+ Arc<PausingObserveAuthority>,
+ Mount,
+ yinyang::core::NodeId,
+ yinyang::core::NodeId,
+ tempfile::TempDir,
+ tempfile::TempDir,
+) {
let backend = TestBackend::default();
let inner: Arc<dyn Authority> = Arc::new(
Fs::create(backend.operator(), BackendProfile::Minio)
@@ -334,77 +352,124 @@ async fn independent_node_opens_do_not_share_the_namespace_lock() {
let mount = Mount::open(authority, mount_temp.path(), false)
.await
.unwrap();
+ (pausing, mount, a, b, setup_temp, mount_temp)
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn node_materialization_is_independent_and_broadcasts_completion() {
+ let (pausing, mount, a, b, _setup_temp, _mount_temp) = pausing_mount().await;
pausing.pause_next_observe();
- let a_task = {
+ let owner = {
let mount = mount.clone();
tokio::spawn(async move { mount.open_node(a, false).await })
};
pausing.entered.acquire().await.unwrap().forget();
- let mut b_task = {
+ let b_file = tokio::time::timeout(Duration::from_secs(5), async {
let mount = mount.clone();
- tokio::spawn(async move { mount.open_node(b, false).await })
- };
- let b_file = tokio::time::timeout(Duration::from_secs(5), &mut b_task)
- .await
- .expect("opening an independent node waited for another node's remote observation")
- .unwrap()
- .unwrap();
+ mount.open_node(b, false).await
+ })
+ .await
+ .expect("opening an independent node waited for another node's remote observation")
+ .unwrap();
drop(b_file);
- let a_second = {
- let mount = mount.clone();
- tokio::spawn(async move { mount.open_node(a, false).await })
- };
+ let waiters: Vec<_> = (0..2)
+ .map(|_| {
+ let mount = mount.clone();
+ tokio::spawn(async move { mount.open_node(a, false).await })
+ })
+ .collect();
+ tokio::time::sleep(Duration::from_millis(250)).await;
pausing.release.add_permits(1);
- let a_file = a_task.await.unwrap().unwrap();
- let a_second = a_second.await.unwrap().unwrap();
- assert_eq!(
- a_file.status().await.unwrap().id,
- a_second.status().await.unwrap().id
- );
- drop(a_second);
+ let a_file = tokio::time::timeout(Duration::from_secs(5), owner)
+ .await
+ .expect("first open did not complete after the observation was released")
+ .unwrap()
+ .unwrap();
+ let first_id = a_file.status().await.unwrap().id;
+ for waiter in waiters {
+ let file = tokio::time::timeout(Duration::from_secs(5), waiter)
+ .await
+ .expect("a same-node waiter was not broadcast awake")
+ .unwrap()
+ .unwrap();
+ assert_eq!(file.status().await.unwrap().id, first_id);
+ }
drop(a_file);
- assert!(mount.node_status(a).await.unwrap().is_some());
- assert!(mount.node_status(b).await.unwrap().is_some());
+ assert_eq!(mount.status().await.unwrap().handles.len(), 2);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn cancelled_first_open_releases_the_node_gate() {
- let backend = TestBackend::default();
- let inner: Arc<dyn Authority> = Arc::new(
- Fs::create(backend.operator(), BackendProfile::Minio)
- .await
- .unwrap(),
- );
- let setup_temp = tempfile::tempdir().unwrap();
- let setup = Runtime::open(inner.clone(), setup_temp.path())
+async fn failed_first_open_broadcasts_retry_to_all_waiters() {
+ let (pausing, mount, a, _b, _setup_temp, _mount_temp) = pausing_mount().await;
+ pausing.pause_next_observe();
+ pausing.fail_next_observe();
+ let owner = {
+ let mount = mount.clone();
+ tokio::spawn(async move { mount.open_node(a, false).await })
+ };
+ pausing.entered.acquire().await.unwrap().forget();
+ let waiters: Vec<_> = (0..2)
+ .map(|_| {
+ let mount = mount.clone();
+ tokio::spawn(async move { mount.open_node(a, false).await })
+ })
+ .collect();
+ tokio::time::sleep(Duration::from_millis(250)).await;
+ pausing.release.add_permits(1);
+ let error = tokio::time::timeout(Duration::from_secs(5), owner)
.await
- .unwrap();
- let (a, _) = setup
- .create_file_with_node(inner.root(), "a", false)
+ .expect("failed owner did not finish")
+ .unwrap()
+ .unwrap_err();
+ assert_eq!(error.kind(), ErrorKind::Storage);
+ let mut waiters = waiters.into_iter();
+ let first = tokio::time::timeout(Duration::from_secs(5), waiters.next().unwrap())
.await
+ .expect("first waiter did not retry after owner failure")
+ .unwrap()
.unwrap();
- drop(setup);
-
- let pausing = Arc::new(PausingObserveAuthority::new(inner));
- let authority: Arc<dyn Authority> = pausing.clone();
- let mount_temp = tempfile::tempdir().unwrap();
- let mount = Mount::open(authority, mount_temp.path(), false)
+ let first_id = first.status().await.unwrap().id;
+ let second = tokio::time::timeout(Duration::from_secs(5), waiters.next().unwrap())
.await
+ .expect("second waiter did not retry after owner failure")
+ .unwrap()
.unwrap();
+ assert_eq!(second.status().await.unwrap().id, first_id);
+ assert_eq!(mount.status().await.unwrap().handles.len(), 1);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn cancelled_first_open_broadcasts_retry_to_all_waiters() {
+ let (pausing, mount, a, _b, _setup_temp, _mount_temp) = pausing_mount().await;
pausing.pause_next_observe();
let cancelled = {
let mount = mount.clone();
tokio::spawn(async move { mount.open_node(a, false).await })
};
pausing.entered.acquire().await.unwrap().forget();
+ let waiters: Vec<_> = (0..2)
+ .map(|_| {
+ let mount = mount.clone();
+ tokio::spawn(async move { mount.open_node(a, false).await })
+ })
+ .collect();
+ tokio::time::sleep(Duration::from_millis(250)).await;
cancelled.abort();
assert!(cancelled.await.unwrap_err().is_cancelled());
- let file = tokio::time::timeout(Duration::from_secs(5), mount.open_node(a, false))
+ let mut waiters = waiters.into_iter();
+ let first = tokio::time::timeout(Duration::from_secs(5), waiters.next().unwrap())
.await
- .expect("cancelled open left the per-node gate claimed")
+ .expect("first waiter did not retry after cancellation")
+ .unwrap()
.unwrap();
- assert!(mount.node_status(a).await.unwrap().is_some());
- drop(file);
+ let first_id = first.status().await.unwrap().id;
+ let second = tokio::time::timeout(Duration::from_secs(5), waiters.next().unwrap())
+ .await
+ .expect("second waiter did not retry after cancellation")
+ .unwrap()
+ .unwrap();
+ assert_eq!(second.status().await.unwrap().id, first_id);
+ assert_eq!(mount.status().await.unwrap().handles.len(), 1);
}