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:44 UTC
(opendal-yinyang) branch xuanwo/linux-fuse-writable updated (8d677f1 -> ba8cd41)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a change to branch xuanwo/linux-fuse-writable
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
from 8d677f1 feat(mount): expose staged metadata and a reportable flush contract
new b284062 fix(mount): release the namespace lock across node materialization
new ba8cd41 fix(mount): broadcast open gate completion and atomic gate registration
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
specs/mount-runtime.md | 13 ++-
src/mount.rs | 183 +++++++++++++++++++++++++++-----------
tests/mount.rs | 237 ++++++++++++++++++++++++++++++++++++++++++++++++-
3 files changed, 380 insertions(+), 53 deletions(-)
(opendal-yinyang) 01/02: fix(mount): release the namespace lock across node materialization
Posted by xu...@apache.org.
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 b2840620dd1b1cbb81afb4f182fba811a1a86562
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Oct 26 15:29:19 2026 +0800
fix(mount): release the namespace lock across node materialization
First opens now serialize per NodeId through a short-lived gate while the instance node map is held only for lookup and accounting, so opening one node cannot block an independent node's remote materialization. NodeGuard keeps refresh, fsync and status from racing reclaim. Deterministic tests pause one observe and prove another node opens independently, concurrent first opens share one handle, and cancellation releases the gate.
Co-authored-by: Chojan Shang <ps...@apache.org>
---
specs/mount-runtime.md | 9 ++-
src/mount.rs | 132 ++++++++++++++++++++++++++++---------
tests/mount.rs | 172 ++++++++++++++++++++++++++++++++++++++++++++++++-
3 files changed, 279 insertions(+), 34 deletions(-)
diff --git a/specs/mount-runtime.md b/specs/mount-runtime.md
index 70ac3a9..105e8fb 100644
--- a/specs/mount-runtime.md
+++ b/specs/mount-runtime.md
@@ -11,8 +11,13 @@ Do not mix private library handles into this staging directory.
writes from other references immediately after local acceptance. Writes,
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, refresh and reclaim serialize through the instance node map.
-Both read-only and writable references currently use full-file staging.
+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
+local open is never exposed and is discarded when staging is reopened. Both
+read-only and writable references currently use full-file staging.
Dropping a reference performs no I/O and never publishes or deletes data.
`reclaim` rejects referenced, pending, frozen or failed nodes. Clean unreferenced
diff --git a/src/mount.rs b/src/mount.rs
index 58baa9f..1aabd54 100644
--- a/src/mount.rs
+++ b/src/mount.rs
@@ -19,19 +19,47 @@
use crate::runtime::{Error, ErrorKind, FileHandle, HandleStatus, Result, Runtime, RuntimeStatus};
use std::collections::BTreeMap;
use std::path::Path;
-use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
-use tokio::sync::Mutex;
+use std::sync::{Arc, Mutex as StdMutex};
+use tokio::sync::{Mutex, Semaphore};
use yinyang_core::{Authority, NodeId, Receipt};
struct Node {
handle: Mutex<FileHandle>,
references: AtomicUsize,
}
+struct OpenGate {
+ inner: Arc<Inner>,
+ id: NodeId,
+ ready: Arc<Semaphore>,
+}
+impl Drop for OpenGate {
+ fn drop(&mut self) {
+ if let Ok(mut opening) = self.inner.opening.lock() {
+ opening.remove(&self.id);
+ }
+ self.ready.add_permits(1);
+ }
+}
+struct NodeGuard {
+ node: Arc<Node>,
+}
+impl NodeGuard {
+ fn new(node: Arc<Node>) -> Self {
+ node.references.fetch_add(1, Ordering::Relaxed);
+ Self { node }
+ }
+}
+impl Drop for NodeGuard {
+ fn drop(&mut self) {
+ self.node.references.fetch_sub(1, Ordering::Relaxed);
+ }
+}
struct Inner {
runtime: Runtime,
read_only: bool,
nodes: Mutex<BTreeMap<NodeId, Arc<Node>>>,
+ opening: StdMutex<BTreeMap<NodeId, Arc<Semaphore>>>,
}
/// A mount instance owns one recoverable staged version per node, not per open.
/// It does not install a mount or manage kernel caches.
@@ -76,6 +104,7 @@ impl Mount {
runtime,
read_only,
nodes: Mutex::new(nodes),
+ opening: StdMutex::new(BTreeMap::new()),
}),
})
}
@@ -88,14 +117,14 @@ impl Mount {
/// Observe staged metadata for one shared node without opening a reference.
/// `None` means this mount has no shared handle for the node.
pub async fn node_status(&self, id: NodeId) -> Result<Option<HandleStatus>> {
- let node = {
+ let guard = {
let nodes = self.inner.nodes.lock().await;
- nodes.get(&id).cloned()
+ nodes.get(&id).cloned().map(NodeGuard::new)
};
- let Some(node) = node else {
+ let Some(guard) = guard else {
return Ok(None);
};
- node.handle.lock().await.status().await.map(Some)
+ guard.node.handle.lock().await.status().await.map(Some)
}
/// Namespace operations use the same conditional runtime contract. Do not
/// open private FileHandles through this runtime alongside Mount references.
@@ -106,37 +135,75 @@ impl Mount {
if writable && self.inner.read_only {
return Err(Error::State(ErrorKind::ReadOnly, "mount is read-only"));
}
- // Serialize first materialization with opens and reclaim. File I/O and
- // publication use per-node locks, never this namespace lock.
- let mut nodes = self.inner.nodes.lock().await;
- let node = match nodes.get(&id) {
- Some(node) => node.clone(),
- None => {
- let handle = self
- .inner
- .runtime
- .open_node(id, !self.inner.read_only)
- .await?;
- let node = Arc::new(Node {
- handle: Mutex::new(handle),
- references: AtomicUsize::new(0),
- });
+ loop {
+ let existing = {
+ let nodes = self.inner.nodes.lock().await;
+ nodes.get(&id).map(|node| {
+ node.references.fetch_add(1, Ordering::Relaxed);
+ node.clone()
+ })
+ };
+ 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)
+ }
+ }
+ };
+ if waiting {
+ ready
+ .acquire()
+ .await
+ .map_err(|_| Error::local("mount open coordination closed"))?
+ .forget();
+ continue;
+ }
+
+ // Only one task materializes a NodeId. The namespace map is not
+ // held across the runtime's remote observation and staging read.
+ let gate = OpenGate {
+ inner: self.inner.clone(),
+ id,
+ ready,
+ };
+ let handle = self
+ .inner
+ .runtime
+ .open_node(id, !self.inner.read_only)
+ .await?;
+ let node = Arc::new(Node {
+ handle: Mutex::new(handle),
+ references: AtomicUsize::new(0),
+ });
+ {
+ let mut nodes = self.inner.nodes.lock().await;
nodes.insert(id, node.clone());
- node
+ node.references.fetch_add(1, Ordering::Relaxed);
}
- };
- node.references.fetch_add(1, Ordering::Relaxed);
- Ok(MountFile { node, writable })
+ drop(gate);
+ return Ok(MountFile { node, writable });
+ }
}
/// Replace a clean local version with a fresh remote observation. Pending,
/// frozen or failed versions are never rebased. Adapters must coordinate any
/// OS cache invalidation outside this call; no OS callbacks run under locks.
pub async fn refresh(&self, id: NodeId) -> Result<()> {
- let nodes = self.inner.nodes.lock().await;
- let Some(node) = nodes.get(&id) else {
+ let guard = {
+ let nodes = self.inner.nodes.lock().await;
+ nodes.get(&id).cloned().map(NodeGuard::new)
+ };
+ let Some(guard) = guard else {
return Ok(());
};
- let mut handle = node.handle.lock().await;
+ let mut handle = guard.node.handle.lock().await;
let state = handle.status().await?;
if state.pending || state.frozen || state.error.is_some() {
return Err(Error::State(
@@ -157,10 +224,13 @@ impl Mount {
/// Persist every currently staged node remotely. Continue after one failure
/// so independent nodes can finish; the original failures remain in status.
pub async fn fsync(&self) -> Result<()> {
- let nodes: Vec<_> = self.inner.nodes.lock().await.values().cloned().collect();
+ let nodes: Vec<_> = {
+ let nodes = self.inner.nodes.lock().await;
+ nodes.values().cloned().map(NodeGuard::new).collect()
+ };
let mut failure = None;
- for node in nodes {
- if let Err(error) = node.handle.lock().await.fsync().await {
+ for node in &nodes {
+ if let Err(error) = node.node.handle.lock().await.fsync().await {
failure.get_or_insert(error);
}
}
diff --git a/tests/mount.rs b/tests/mount.rs
index 338a786..7b05420 100644
--- a/tests/mount.rs
+++ b/tests/mount.rs
@@ -18,12 +18,79 @@
#[path = "../crates/core/tests/support/mod.rs"]
mod support;
use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use std::{future::Future, pin::Pin};
use support::TestBackend;
+use tokio::sync::Semaphore;
use yinyang::core::service::{MetadataService, ServiceClient, serve};
-use yinyang::core::{Authority, BackendProfile, Fs};
+use yinyang::core::{
+ Authority, BackendProfile, CommitOutcome, DataStore, Fs, NodeId, Revision, Snapshot,
+ Transaction,
+};
use yinyang::mount::Mount;
use yinyang::runtime::{ErrorKind, Runtime};
+struct PausingObserveAuthority {
+ inner: Arc<dyn Authority>,
+ entered: Semaphore,
+ release: Semaphore,
+ pause_next_observe: AtomicBool,
+}
+impl PausingObserveAuthority {
+ fn new(inner: Arc<dyn Authority>) -> Self {
+ Self {
+ inner,
+ entered: Semaphore::new(0),
+ release: Semaphore::new(0),
+ pause_next_observe: AtomicBool::new(false),
+ }
+ }
+ fn pause_next_observe(&self) {
+ self.pause_next_observe.store(true, Ordering::SeqCst);
+ }
+}
+impl std::fmt::Debug for PausingObserveAuthority {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("PausingObserveAuthority")
+ .finish_non_exhaustive()
+ }
+}
+impl Authority for PausingObserveAuthority {
+ fn data(&self) -> &DataStore {
+ self.inner.data()
+ }
+ fn filesystem(&self) -> NodeId {
+ self.inner.filesystem()
+ }
+ fn root(&self) -> NodeId {
+ self.inner.root()
+ }
+ fn observe_latest(
+ &self,
+ ) -> Pin<Box<dyn Future<Output = yinyang::core::Result<Snapshot>> + Send + '_>> {
+ Box::pin(async move {
+ if self.pause_next_observe.swap(false, Ordering::SeqCst) {
+ self.entered.add_permits(1);
+ self.release.acquire().await.unwrap().forget();
+ }
+ self.inner.observe_latest().await
+ })
+ }
+ fn observe_revision(
+ &self,
+ revision: Revision,
+ ) -> Pin<Box<dyn Future<Output = yinyang::core::Result<Snapshot>> + Send + '_>> {
+ self.inner.observe_revision(revision)
+ }
+ fn commit_batch<'a>(
+ &'a self,
+ requests: &'a [Transaction],
+ ) -> Pin<Box<dyn Future<Output = yinyang::core::Result<Vec<CommitOutcome>>> + Send + 'a>> {
+ self.inner.commit_batch(requests)
+ }
+}
+
async fn exercise(authority: Arc<dyn Authority>) {
let temp = tempfile::tempdir().unwrap();
let mount = Mount::open(authority.clone(), temp.path(), false)
@@ -238,3 +305,106 @@ async fn flush_publishes_but_reference_release_does_not() {
mount.reclaim(pending).await.unwrap();
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() {
+ 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())
+ .await
+ .unwrap();
+ let (a, _) = setup
+ .create_file_with_node(inner.root(), "a", false)
+ .await
+ .unwrap();
+ let (b, _) = setup
+ .create_file_with_node(inner.root(), "b", false)
+ .await
+ .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)
+ .await
+ .unwrap();
+ pausing.pause_next_observe();
+ let a_task = {
+ 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 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();
+ drop(b_file);
+ let a_second = {
+ let mount = mount.clone();
+ tokio::spawn(async move { mount.open_node(a, false).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);
+ drop(a_file);
+ assert!(mount.node_status(a).await.unwrap().is_some());
+ assert!(mount.node_status(b).await.unwrap().is_some());
+}
+
+#[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())
+ .await
+ .unwrap();
+ let (a, _) = setup
+ .create_file_with_node(inner.root(), "a", false)
+ .await
+ .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)
+ .await
+ .unwrap();
+ 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();
+ cancelled.abort();
+ assert!(cancelled.await.unwrap_err().is_cancelled());
+
+ let file = tokio::time::timeout(Duration::from_secs(5), mount.open_node(a, false))
+ .await
+ .expect("cancelled open left the per-node gate claimed")
+ .unwrap();
+ assert!(mount.node_status(a).await.unwrap().is_some());
+ drop(file);
+}
(opendal-yinyang) 02/02: fix(mount): broadcast open gate completion and atomic gate registration
Posted by xu...@apache.org.
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);
}