perf(virtiofs): batch non-DAX writeback (#2122)

* perf(virtiofs): batch non-DAX writeback

Buffer writes in the generic page cache only after FUSE_WRITEBACK_CACHE is negotiated, then claim and submit bounded contiguous batches according to the negotiated write and page limits.

Preserve Linux-compatible fsync, close, truncate, mmap, invalidation, stable EOF, redirty, short-write, and errseq behavior. Separate terminal writeback completion from generic page-cache workers so host invalidation cannot strand published Writeback pages behind its own waiters.

Harden kernel-thread creation and wakeup ordering required by the new worker pools. Add exact FUSE/page-cache counters, benchmark phase reporting, focused FUSE regressions, and root-only kthread and completion-domain selftests.

Validated with make kernel -j2, kernel formatting and diff checks, the completion-domain selftest, targeted close/flush coverage, and FuseExtended 69/69 in a DragonOS guest.

Signed-off-by: longjin <longjin@dragonos.org>

* refactor(fuse): group negotiated io limits

Pass negotiated read, write, page, capability, and effective payload limits through a dedicated stats value object.

This keeps the INIT statistics update cohesive and satisfies the project-wide Clippy argument-count lint enforced by make fmt without changing the negotiated values or publication ordering.

Signed-off-by: longjin <longjin@dragonos.org>

---------

Signed-off-by: longjin <longjin@dragonos.org>

kernel/src/arch/riscv64/process/kthread.rs

@@ -58,10 +58,6 @@ impl KernelThreadMechanism {
             e
         })?;
 
-        let pcb = ProcessManager::find(pid).unwrap();
-        pcb.set_name(info.name().clone());
-        info.setup_pcb(&pcb);
-
         return Ok(pid);
     }
 }

kernel/src/arch/x86_64/process/kthread.rs

@@ -59,10 +59,6 @@ impl KernelThreadMechanism {
             unsafe { KernelThreadCreateInfo::parse_unsafe_arc_ptr(create_info) };
         })?;
 
-        let pcb = ProcessManager::find_task_by_vpid(pid).unwrap();
-        pcb.set_name(info.name().clone());
-        info.setup_pcb(&pcb);
-
         return Ok(pid);
     }
 }

kernel/src/debug/kthread.rs

@@ -0,0 +1,75 @@
+use alloc::{string::String, string::ToString};
+
+use system_error::SystemError;
+
+use crate::{
+    debug::sysfs::debugfs_kobj,
+    driver::base::kobject::KObject,
+    filesystem::{
+        kernfs::callback::{KernCallbackData, KernFSCallback, KernFilePrivateData},
+        vfs::{InodeMode, PollStatus},
+    },
+};
+
+#[derive(Debug)]
+struct KthreadSelftestCallback;
+
+impl KernFSCallback for KthreadSelftestCallback {
+    fn open(&self, mut data: KernCallbackData) -> Result<(), SystemError> {
+        let report = crate::process::kthread::run_debug_selftests()?;
+        data.file_private_data_mut()
+            .replace(KernFilePrivateData::DebugTextSnapshot(report));
+        Ok(())
+    }
+
+    fn read(
+        &self,
+        data: KernCallbackData,
+        buf: &mut [u8],
+        offset: usize,
+    ) -> Result<usize, SystemError> {
+        let report: &String = match data.file_private_data() {
+            Some(KernFilePrivateData::DebugTextSnapshot(report)) => report,
+            _ => return Err(SystemError::EINVAL),
+        };
+        let bytes = report.as_bytes();
+        if offset >= bytes.len() {
+            return Ok(0);
+        }
+        let len = buf.len().min(bytes.len() - offset);
+        buf[..len].copy_from_slice(&bytes[offset..offset + len]);
+        Ok(len)
+    }
+
+    fn write(
+        &self,
+        _data: KernCallbackData,
+        _buf: &[u8],
+        _offset: usize,
+    ) -> Result<usize, SystemError> {
+        Err(SystemError::EPERM)
+    }
+
+    fn poll(&self, _data: KernCallbackData) -> Result<PollStatus, SystemError> {
+        Ok(PollStatus::READ)
+    }
+}
+
+pub fn init_debugfs_kthread() -> Result<(), SystemError> {
+    let debugfs = debugfs_kobj();
+    let root = debugfs.inode().ok_or(SystemError::ENOENT)?;
+    let kthread = root.add_dir(
+        "kthread".to_string(),
+        InodeMode::from_bits_truncate(0o555),
+        None,
+        None,
+    )?;
+    kthread.add_file(
+        "selftest".to_string(),
+        InodeMode::S_IRUSR,
+        Some(4096),
+        None,
+        Some(&KthreadSelftestCallback),
+    )?;
+    Ok(())
+}

kernel/src/debug/mod.rs

@@ -4,6 +4,8 @@ pub mod fuse;
 pub mod jump_label;
 pub mod klog;
 pub mod kprobe;
+pub mod kthread;
+pub mod page_cache;
 pub mod panic;
 pub mod rcu;
 pub mod sysfs;

kernel/src/debug/page_cache.rs

@@ -0,0 +1,75 @@
+use alloc::{string::String, string::ToString};
+
+use system_error::SystemError;
+
+use crate::{
+    debug::sysfs::debugfs_kobj,
+    driver::base::kobject::KObject,
+    filesystem::{
+        kernfs::callback::{KernCallbackData, KernFSCallback, KernFilePrivateData},
+        vfs::{InodeMode, PollStatus},
+    },
+};
+
+#[derive(Debug)]
+struct PageCacheCompletionSelftestCallback;
+
+impl KernFSCallback for PageCacheCompletionSelftestCallback {
+    fn open(&self, mut data: KernCallbackData) -> Result<(), SystemError> {
+        let report = crate::filesystem::page_cache::run_completion_domain_debug_selftest()?;
+        data.file_private_data_mut()
+            .replace(KernFilePrivateData::DebugTextSnapshot(report));
+        Ok(())
+    }
+
+    fn read(
+        &self,
+        data: KernCallbackData,
+        buf: &mut [u8],
+        offset: usize,
+    ) -> Result<usize, SystemError> {
+        let report: &String = match data.file_private_data() {
+            Some(KernFilePrivateData::DebugTextSnapshot(report)) => report,
+            _ => return Err(SystemError::EINVAL),
+        };
+        let bytes = report.as_bytes();
+        if offset >= bytes.len() {
+            return Ok(0);
+        }
+        let len = buf.len().min(bytes.len() - offset);
+        buf[..len].copy_from_slice(&bytes[offset..offset + len]);
+        Ok(len)
+    }
+
+    fn write(
+        &self,
+        _data: KernCallbackData,
+        _buf: &[u8],
+        _offset: usize,
+    ) -> Result<usize, SystemError> {
+        Err(SystemError::EPERM)
+    }
+
+    fn poll(&self, _data: KernCallbackData) -> Result<PollStatus, SystemError> {
+        Ok(PollStatus::READ)
+    }
+}
+
+pub fn init_debugfs_page_cache() -> Result<(), SystemError> {
+    let debugfs = debugfs_kobj();
+    let root = debugfs.inode().ok_or(SystemError::ENOENT)?;
+    let page_cache = root.add_dir(
+        "page_cache".to_string(),
+        InodeMode::from_bits_truncate(0o555),
+        None,
+        None,
+    )?;
+    page_cache.add_file(
+        "completion_domain_selftest".to_string(),
+        InodeMode::S_IRUSR,
+        Some(4096),
+        None,
+        Some(&PageCacheCompletionSelftestCallback),
+    )?;
+    Ok(())
+}

kernel/src/debug/sysfs/mod.rs

@@ -25,6 +25,8 @@ fn debugfs_init() -> Result<(), SystemError> {
     super::errseq::init_debugfs_errseq()?;
     super::ext4::init_debugfs_ext4()?;
     super::fuse::init_debugfs_fuse()?;
+    super::kthread::init_debugfs_kthread()?;
+    super::page_cache::init_debugfs_page_cache()?;
     return Ok(());
 }
 

kernel/src/filesystem/fuse/conn/daemon.rs

@@ -4,7 +4,11 @@ use core::{mem::size_of, sync::atomic::Ordering};
 use num_traits::FromPrimitive;
 use system_error::SystemError;
 
-use crate::filesystem::epoll::{EPollEventType, EPollItem};
+use crate::{
+    arch::MMArch,
+    filesystem::epoll::{EPollEventType, EPollItem},
+    mm::MemoryManagementArch,
+};
 
 use super::super::protocol::{
     fuse_read_struct, FuseAttrOut, FuseEntryOut, FuseInHeader, FuseInitOut, FuseNotifyDeleteOut,
@@ -15,7 +19,7 @@ use super::super::protocol::{
     FUSE_MAP_ALIGNMENT, FUSE_MAX_PAGES, FUSE_MIN_READ_BUFFER, FUSE_MKDIR, FUSE_MKNOD,
     FUSE_NOTIFY_DELETE, FUSE_NOTIFY_INVAL_ENTRY, FUSE_NOTIFY_INVAL_INODE, FUSE_NOTIFY_POLL,
     FUSE_NOTIFY_RETRIEVE, FUSE_NOTIFY_STORE, FUSE_READ, FUSE_REMOVEXATTR, FUSE_SETATTR,
-    FUSE_SETXATTR, FUSE_STATFS, FUSE_SYMLINK,
+    FUSE_SETXATTR, FUSE_STATFS, FUSE_SYMLINK, FUSE_WRITEBACK_CACHE,
 };
 use super::{
     stats, trace, wait_with_recheck, FuseConn, FuseConnInner, FuseInitNegotiated, FuseRequest,
@@ -615,13 +619,23 @@ impl FuseConn {
             })?;
             self.background
                 .configure(max_background, congestion_threshold);
-            stats::on_fuse_read_limits_negotiated(
-                self.max_read(),
-                negotiated_max_pages as usize,
-                init_out.max_readahead as usize,
-                (enabled_flags & FUSE_ASYNC_READ) != 0,
-                self.effective_read_payload_limit(),
-            );
+            stats::on_fuse_io_limits_negotiated(stats::NegotiatedFuseIoLimits {
+                max_read: self.max_read(),
+                max_write: self.max_write(),
+                max_pages: negotiated_max_pages as usize,
+                max_readahead: init_out.max_readahead as usize,
+                async_read: (enabled_flags & FUSE_ASYNC_READ) != 0,
+                writeback_cache: (enabled_flags & FUSE_WRITEBACK_CACHE) != 0,
+                effective_read_payload_limit: self.effective_read_payload_limit(),
+                effective_write_payload_limit: core::cmp::min(
+                    core::cmp::min(
+                        negotiated_max_pages as usize,
+                        self.max_write() / MMArch::PAGE_SIZE,
+                    ),
+                    64,
+                )
+                .saturating_mul(MMArch::PAGE_SIZE),
+            });
             self.init_wait.wakeup(None);
         } else {
             self.claim_pending_reply(out_hdr.unique, &pending, |_| {})?;

kernel/src/filesystem/fuse/fs.rs

@@ -31,6 +31,7 @@ use super::{
         fuse_read_struct, FuseStatfsOut, FOPEN_DIRECT_IO, FUSE_ATTR_SUBMOUNT, FUSE_ROOT_ID,
         FUSE_STATFS,
     },
+    stats,
 };
 
 #[derive(Debug)]
@@ -843,11 +844,18 @@ impl FileSystem for FuseFS {
             p.node.clone()
         };
 
-        node.with_writeback_admission(|| {
+        let mut result = VmFaultReason::VM_FAULT_SIGBUS;
+        let admitted = node.try_with_writeback_admission(&mut || {
             let Ok(_pin) = node.pin_writeback_handle() else {
-                return VmFaultReason::VM_FAULT_SIGBUS;
+                return Ok(());
             };
-            let result = PageFaultHandler::filemap_page_mkwrite(pfm);
+            let Some(metadata) = node.cached_metadata_snapshot() else {
+                return Ok(());
+            };
+            result = PageFaultHandler::filemap_page_mkwrite_with_stable_size(
+                pfm,
+                metadata.size.max(0) as usize,
+            );
             if !result.intersects(
                 VmFaultReason::VM_FAULT_SIGBUS
                     | VmFaultReason::VM_FAULT_OOM
@@ -855,8 +863,21 @@ impl FileSystem for FuseFS {
             ) {
                 node.note_mmap_write();
             }
-            result
-        })
+            Ok(())
+        });
+        match admitted {
+            Ok(true) => result,
+            Ok(false) => {
+                // Never block on a writer-preferred admission semaphore while
+                // the architecture fault path owns AddressSpace::write(). The
+                // retry waiter acquires/releases it only after the MM guard is
+                // dropped, closing the barrier<->mkclean cycle.
+                stats::on_mmap_writeback_admission_retry();
+                pfm.set_retry_wait(node.writeback_admission_retry_wait());
+                VmFaultReason::VM_FAULT_RETRY
+            }
+            Err(_) => VmFaultReason::VM_FAULT_SIGBUS,
+        }
     }
 
     fn mprotect(&self, _old_vm_flags: VmFlags, new_vm_flags: VmFlags) -> Result<(), SystemError> {

kernel/src/filesystem/fuse/inode.rs

@@ -13,7 +13,6 @@ use core::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering
 
 use system_error::SystemError;
 
-use crate::time::timekeep::ktime_get_real_ns;
 use crate::{
     driver::base::device::device_number::DeviceNumber,
     filesystem::{
@@ -26,7 +25,7 @@ use crate::{
         wait_queue::WaitQueue,
     },
     mm::{fault::FaultRetryWait, MemoryManagementArch},
-    time::PosixTimeSpec,
+    time::{jiffies::NSEC_PER_JIFFY, timer::clock, PosixTimeSpec, NSEC_PER_SEC},
 };
 
 use super::reply::FuseReply;
@@ -300,7 +299,7 @@ pub struct FuseNode {
     dax_mappings: RwSem<DaxMappingTree>,
     dax_host_invalidation: Arc<DaxHostInvalidationGate>,
     dax_pte_epoch: AtomicU64,
-    cached_metadata_deadline_ns: AtomicU64,
+    cached_metadata_deadline_ticks: AtomicU64,
     attr_version: AtomicU64,
     /// Version chain produced while short READ replies from one metadata
     /// snapshot monotonically converge on the lowest observed EOF.
@@ -325,7 +324,7 @@ pub struct FuseNode {
 struct FuseLookupCacheEntry {
     child: Arc<FuseNode>,
     generation: u64,
-    deadline_ns: u64,
+    deadline_ticks: u64,
 }
 
 impl FuseNode {
@@ -377,7 +376,7 @@ impl FuseNode {
             dax_mappings: RwSem::new(DaxMappingTree::default()),
             dax_host_invalidation: DaxHostInvalidationGate::new(),
             dax_pte_epoch: AtomicU64::new(0),
-            cached_metadata_deadline_ns: AtomicU64::new(if has_cached { u64::MAX } else { 0 }),
+            cached_metadata_deadline_ticks: AtomicU64::new(if has_cached { u64::MAX } else { 0 }),
             attr_version: AtomicU64::new(initial_attr_epoch),
             short_read_source_attr_version: AtomicU64::new(0),
             short_read_chain_attr_version: AtomicU64::new(0),
@@ -1233,7 +1232,7 @@ impl FuseNode {
         *metadata = Some(md);
         self.bump_attr_version();
         drop(metadata);
-        self.cached_metadata_deadline_ns
+        self.cached_metadata_deadline_ticks
             .store(Self::cache_deadline(valid, valid_nsec), Ordering::Relaxed);
         self.note_dax_attr_change(attr_flags);
     }
@@ -1271,7 +1270,7 @@ impl FuseNode {
         *metadata = Some(md.clone());
         self.bump_attr_version();
         drop(metadata);
-        self.cached_metadata_deadline_ns.store(
+        self.cached_metadata_deadline_ticks.store(
             if stale_reply {
                 0
             } else {
@@ -1303,7 +1302,8 @@ impl FuseNode {
 
     pub(crate) fn invalidate_cached_metadata(&self) {
         self.bump_attr_version();
-        self.cached_metadata_deadline_ns.store(0, Ordering::Release);
+        self.cached_metadata_deadline_ticks
+            .store(0, Ordering::Release);
     }
 
     /// 累计该 inode 在 userspace daemon 侧持有的 LOOKUP 引用。
@@ -1329,18 +1329,41 @@ impl FuseNode {
         let _ = self.conn.queue_forget(self.nodeid, nlookup);
     }
 
-    fn now_ns() -> u64 {
-        ktime_get_real_ns().max(0) as u64
+    fn now_ticks() -> u64 {
+        // FUSE cache expiry is an elapsed-time deadline, not wall-clock time.
+        // Match Linux fuse_time_to_jiffies(): use the monotonic timer tick
+        // counter so each hot-path attr/entry-cache check is an in-memory read
+        // and settimeofday cannot extend or prematurely expire the cache.
+        clock()
     }
 
-    fn cache_deadline(valid: u64, valid_nsec: u32) -> u64 {
+    fn cache_timeout_ticks(valid: u64, valid_nsec: u32) -> u64 {
         if valid == 0 && valid_nsec == 0 {
             return 0;
         }
-        let delta_ns = valid
-            .saturating_mul(1_000_000_000)
-            .saturating_add(valid_nsec as u64);
-        Self::now_ns().saturating_add(delta_ns)
+
+        // Linux clamps the daemon-provided nanosecond component before
+        // converting the relative timeout to jiffies.  Calculate in u128 so a
+        // malformed, extremely large finite timeout cannot overflow into the
+        // u64::MAX sentinel reserved for kernel-owned permanent snapshots.
+        let nsec = valid_nsec.min(NSEC_PER_SEC - 1) as u128;
+        let delta_ns = (valid as u128)
+            .saturating_mul(NSEC_PER_SEC as u128)
+            .saturating_add(nsec);
+        delta_ns
+            .div_ceil(NSEC_PER_JIFFY as u128)
+            .min((u64::MAX - 1) as u128) as u64
+    }
+
+    fn cache_deadline(valid: u64, valid_nsec: u32) -> u64 {
+        let delta_ticks = Self::cache_timeout_ticks(valid, valid_nsec);
+        if delta_ticks == 0 {
+            return 0;
+        }
+        Self::now_ticks()
+            .checked_add(delta_ticks)
+            .filter(|deadline| *deadline < u64::MAX)
+            .unwrap_or(u64::MAX - 1)
     }
 
     pub(crate) fn conn(&self) -> &Arc<FuseConn> {
@@ -1494,8 +1517,8 @@ impl FuseNode {
         fh: Option<u64>,
     ) -> Result<Metadata, SystemError> {
         if let Some(m) = self.cached_metadata.lock().clone() {
-            let deadline = self.cached_metadata_deadline_ns.load(Ordering::Relaxed);
-            if deadline == u64::MAX || (deadline != 0 && Self::now_ns() < deadline) {
+            let deadline = self.cached_metadata_deadline_ticks.load(Ordering::Relaxed);
+            if deadline == u64::MAX || (deadline != 0 && Self::now_ticks() < deadline) {
                 return Ok(m);
             }
         }
@@ -1520,8 +1543,8 @@ impl FuseNode {
         // for every AUTO_INVAL_DATA read dominates 4 KiB cached I/O. The TTL is
         // independently published, so reuse that snapshot while it remains
         // valid and issue GETATTR_FH only after expiry.
-        let deadline = self.cached_metadata_deadline_ns.load(Ordering::Acquire);
-        if deadline == u64::MAX || (deadline != 0 && Self::now_ns() < deadline) {
+        let deadline = self.cached_metadata_deadline_ticks.load(Ordering::Acquire);
+        if deadline == u64::MAX || (deadline != 0 && Self::now_ticks() < deadline) {
             return Ok(cached);
         }
         self.fetch_attr_with_file_handle(Some(fh))
@@ -1583,6 +1606,28 @@ mod tests {
         assert_eq!(open_file_query.fh, 0x1234_5678);
     }
 
+    #[test]
+    fn cache_timeout_matches_linux_jiffy_and_nsec_rules() {
+        assert_eq!(FuseNode::cache_timeout_ticks(0, 0), 0);
+        assert_eq!(FuseNode::cache_timeout_ticks(0, 1), 1);
+
+        let clamped = FuseNode::cache_timeout_ticks(0, NSEC_PER_SEC - 1);
+        assert_eq!(FuseNode::cache_timeout_ticks(0, u32::MAX), clamped);
+        assert_eq!(
+            FuseNode::cache_timeout_ticks(1, 0),
+            (NSEC_PER_SEC as u64).div_ceil(NSEC_PER_JIFFY as u64)
+        );
+    }
+
+    #[test]
+    fn finite_cache_timeout_never_becomes_permanent_sentinel() {
+        assert_eq!(
+            FuseNode::cache_timeout_ticks(u64::MAX, u32::MAX),
+            u64::MAX - 1
+        );
+        assert_ne!(FuseNode::cache_deadline(u64::MAX, u32::MAX), u64::MAX);
+    }
+
     #[test]
     fn dax_host_invalidation_gate_blocks_until_last_notification_finishes() {
         let gate = DaxHostInvalidationGate::new();

kernel/src/filesystem/fuse/inode/directory.rs

@@ -48,13 +48,13 @@ impl FuseNode {
             self.invalidate_lookup_cache(name);
             return;
         }
-        let deadline_ns = Self::cache_deadline(valid, valid_nsec);
+        let deadline_ticks = Self::cache_deadline(valid, valid_nsec);
         self.prune_lookup_cache();
 
         let mut removed = Vec::new();
         {
             let mut cache = self.lookup_cache.lock();
-            if deadline_ns == 0 || child.dax_dontcache() {
+            if deadline_ticks == 0 || child.dax_dontcache() {
                 if let Some(entry) = cache.remove(name) {
                     removed.push(entry);
                 }
@@ -69,14 +69,14 @@ impl FuseNode {
                 if let Some(entry) = cache.get_mut(name) {
                     if Arc::ptr_eq(&entry.child, child) {
                         entry.generation = generation;
-                        entry.deadline_ns = deadline_ns;
+                        entry.deadline_ticks = deadline_ticks;
                     } else {
                         let old_entry = replace(
                             entry,
                             FuseLookupCacheEntry {
                                 child: child.clone(),
                                 generation,
-                                deadline_ns,
+                                deadline_ticks,
                             },
                         );
                         removed.push(old_entry);
@@ -87,7 +87,7 @@ impl FuseNode {
                         FuseLookupCacheEntry {
                             child: child.clone(),
                             generation,
-                            deadline_ns,
+                            deadline_ticks,
                         },
                     );
                 }
@@ -134,7 +134,7 @@ impl FuseNode {
     pub(crate) fn notify_expire_child(&self, name: &str) -> Result<(), SystemError> {
         let mut cache = self.lookup_cache.lock();
         let entry = cache.get_mut(name).ok_or(SystemError::ENOENT)?;
-        entry.deadline_ns = 0;
+        entry.deadline_ticks = 0;
         self.invalidate_cached_metadata();
         Ok(())
     }
@@ -172,7 +172,7 @@ impl FuseNode {
         entry: &FuseLookupCacheEntry,
         now: u64,
     ) -> bool {
-        (entry.deadline_ns != u64::MAX && now >= entry.deadline_ns)
+        (entry.deadline_ticks != u64::MAX && now >= entry.deadline_ticks)
             || entry.child.dax_dontcache()
             || entry.child.check_not_stale().is_err()
             || entry.child.generation() != entry.generation
@@ -190,7 +190,7 @@ impl FuseNode {
     }
 
     fn prune_lookup_cache(&self) {
-        let now = Self::now_ns();
+        let now = Self::now_ticks();
         let removed = {
             let mut cache = self.lookup_cache.lock();
             let stale_keys: Vec<String> = cache

kernel/src/filesystem/fuse/inode/file.rs

@@ -15,14 +15,27 @@ use crate::{
         },
         vfs::{file::FileFlags, FilePrivateData, FileType, IndexNode, Metadata, SetMetadataMask},
     },
-    libs::mutex::Mutex,
+    libs::{mutex::Mutex, rwsem::RwSemWriteGuard},
     mm::{
+        fault::FaultRetryWait,
         readahead::{FileReadaheadState, ReadaheadControl},
         MemoryManagementArch,
     },
     time::PosixTimeSpec,
 };
 
+#[derive(Debug)]
+struct FuseWritebackAdmissionRetryWait {
+    node: Arc<FuseNode>,
+}
+
+impl FaultRetryWait for FuseWritebackAdmissionRetryWait {
+    fn wait(&self) -> Result<(), SystemError> {
+        drop(self.node.writeback_barrier.read());
+        Ok(())
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum FuseReadaheadEvent {
     SyncMissAtExpected,
@@ -210,6 +223,55 @@ impl PageCacheBackend for FusePageCacheBackend {
         node.writeback_page_with_handle(index, buf)
     }
 
+    fn write_batch_pages(&self) -> Result<usize, SystemError> {
+        let node = self.node.upgrade().ok_or(SystemError::EIO)?;
+        let max_pages = node.conn().max_pages();
+        let max_write = node.conn().max_write();
+        if max_pages == 0 || max_write < MMArch::PAGE_SIZE {
+            return Err(SystemError::EIO);
+        }
+        let _ = max_pages
+            .checked_mul(MMArch::PAGE_SIZE)
+            .ok_or(SystemError::EOVERFLOW)?;
+        let pages = core::cmp::min(max_pages, max_write / MMArch::PAGE_SIZE);
+        if pages == 0 {
+            return Err(SystemError::EIO);
+        }
+        Ok(pages)
+    }
+
+    fn write_pages(&self, start_index: usize, data: &[u8]) -> Result<(), SystemError> {
+        let node = self.node.upgrade().ok_or(SystemError::EIO)?;
+        match node.writeback_pages_with_handle(start_index, data) {
+            Ok(written) if written == data.len() => Ok(()),
+            Ok(_) => Err(SystemError::EIO),
+            Err(error) => Err(error),
+        }
+    }
+
+    fn with_write_admission(
+        &self,
+        claim: &mut dyn FnMut() -> Result<(), SystemError>,
+    ) -> Result<(), SystemError> {
+        let node = self.node.upgrade().ok_or(SystemError::EIO)?;
+        node.with_writeback_admission(claim)
+    }
+
+    fn try_with_write_admission(
+        &self,
+        claim: &mut dyn FnMut() -> Result<(), SystemError>,
+    ) -> Result<bool, SystemError> {
+        let node = self.node.upgrade().ok_or(SystemError::EIO)?;
+        node.try_with_writeback_admission(claim)
+    }
+
+    fn stable_writeback_size(&self, _inode: &Arc<dyn IndexNode>) -> Result<usize, SystemError> {
+        let node = self.node.upgrade().ok_or(SystemError::EIO)?;
+        node.cached_metadata_snapshot()
+            .ok_or(SystemError::EIO)
+            .map(|metadata| metadata.size.max(0) as usize)
+    }
+
     fn npages(&self) -> usize {
         let Some(node) = self.node.upgrade() else {
             return 0;
@@ -296,25 +358,39 @@ impl FuseNode {
                 // invalidating the ordinary page cache. The host-invalidation
                 // blocker independently prevents DAX window use/PTE publish.
                 let layout = node.dax_layout_write();
-                if let Some(cache) = page_cache.as_ref() {
+                let mut first_error = None;
+                if page_cache.is_some() {
                     // Match Linux invalidate_inode_pages2_range(): only the
                     // notified pages participate in laundering. An unrelated
                     // dirty-page error must not abort the FUSE connection.
-                    cache
-                        .manager()
-                        .launder_range_for_invalidate(start_index, end_index)?;
+                    if let Err(error) =
+                        node.launder_cached_range_admitted(start_index, end_index, &layout)
+                    {
+                        first_error = Some(error);
+                    }
                 }
-                node.invalidate_page_cache_range(
+                // Even when laundering retained an errored/dirty page, evict
+                // every page that was successfully cleaned. Otherwise a late
+                // batch error could leave earlier clean cache entries stale
+                // after the host notification has been acknowledged.
+                if let Err(error) = node.invalidate_page_cache_range(
                     start,
                     end_exclusive
                         .and_then(|end| end.checked_sub(start))
                         .unwrap_or(usize::MAX - start),
-                )?;
+                ) {
+                    if first_error.is_none() {
+                        first_error = Some(error);
+                    }
+                }
                 drop(layout);
 
                 if blocker.lock().is_some() {
                     drop(node.dax_layout_write_for_host_invalidation(start, end_exclusive)?);
                 }
+                if let Some(error) = first_error {
+                    return Err(error);
+                }
                 Ok::<(), SystemError>(())
             })();
             if let Err(error) = result {
@@ -352,6 +428,26 @@ impl FuseNode {
         f()
     }
 
+    pub(crate) fn try_with_writeback_admission(
+        &self,
+        f: &mut dyn FnMut() -> Result<(), SystemError>,
+    ) -> Result<bool, SystemError> {
+        let Some(_guard) = self.writeback_barrier.try_read() else {
+            return Ok(false);
+        };
+        f()?;
+        Ok(true)
+    }
+
+    pub(crate) fn writeback_admission_retry_wait(&self) -> Arc<dyn FaultRetryWait> {
+        Arc::new(FuseWritebackAdmissionRetryWait {
+            node: self
+                .self_ref
+                .upgrade()
+                .expect("live FuseNode must retain its self reference"),
+        })
+    }
+
     pub(crate) fn note_mmap_write(&self) {
         if !self.conn().has_init_flag(FUSE_WRITEBACK_CACHE) {
             return;
@@ -540,7 +636,7 @@ impl FuseNode {
         } else {
             self.writeback_barrier.write()
         };
-        self.sync_dirty_cached_pages()?;
+        self.sync_dirty_cached_pages_admitted(&_barrier)?;
         let mut valid = FATTR_SIZE;
         if lock_owner.is_some() {
             valid |= FATTR_LOCKOWNER;
@@ -582,8 +678,8 @@ impl FuseNode {
                 atime = metadata.atime.tv_sec as u64;
                 atimensec = metadata.atime.tv_nsec as u32;
             }
-            // DragonOS does not currently negotiate WRITEBACK_CACHE. Match
-            // Linux FUSE by accepting the daemon-returned cmtime for truncate
+            // Under WRITEBACK_CACHE, Linux trusts the locally maintained cmtime;
+            // otherwise accept the daemon-returned values for automatic truncate
             // instead of sending a second SETATTR without the file handle.
             if mask.contains(SetMetadataMask::MTIME) && (!automatic || trust_local_cmtime) {
                 valid |= FATTR_MTIME;
@@ -671,7 +767,7 @@ impl FuseNode {
                         .store(chain_version, Ordering::Release);
                     self.pending_short_read_eof
                         .fetch_min(eof as u64, Ordering::AcqRel);
-                    self.cached_metadata_deadline_ns
+                    self.cached_metadata_deadline_ticks
                         .store(u64::MAX, Ordering::Relaxed);
                     if let Some(cache) = self.cached_page_cache() {
                         let start_page = eof / MMArch::PAGE_SIZE;
@@ -797,6 +893,60 @@ impl FuseNode {
         page_cache.manager().sync()
     }
 
+    fn stable_writeback_size(&self) -> Result<usize, SystemError> {
+        Ok(self
+            .cached_metadata_snapshot()
+            .ok_or(SystemError::EIO)?
+            .size
+            .max(0) as usize)
+    }
+
+    pub(super) fn sync_dirty_cached_pages_admitted(
+        &self,
+        _guard: &RwSemWriteGuard<'_, ()>,
+    ) -> Result<(), SystemError> {
+        let Some(page_cache) = self.cached_page_cache() else {
+            return Ok(());
+        };
+        page_cache
+            .manager()
+            .sync_with_stable_size(self.stable_writeback_size()?)
+    }
+
+    pub(super) fn sync_cached_range_admitted(
+        &self,
+        start_index: usize,
+        end_index: usize,
+        _guard: &RwSemWriteGuard<'_, ()>,
+    ) -> Result<(), SystemError> {
+        let Some(page_cache) = self.cached_page_cache() else {
+            return Ok(());
+        };
+        page_cache.manager().sync_range_with_stable_size(
+            start_index,
+            end_index,
+            self.stable_writeback_size()?,
+        )
+    }
+
+    pub(super) fn launder_cached_range_admitted(
+        &self,
+        start_index: usize,
+        end_index: usize,
+        _guard: &RwSemWriteGuard<'_, ()>,
+    ) -> Result<(), SystemError> {
+        let Some(page_cache) = self.cached_page_cache() else {
+            return Ok(());
+        };
+        page_cache
+            .manager()
+            .launder_range_for_invalidate_with_stable_size(
+                start_index,
+                end_index,
+                self.stable_writeback_size()?,
+            )
+    }
+
     pub(super) fn check_and_advance_open_wb_error(
         &self,
         data: &FuseOpenPrivateData,
@@ -816,22 +966,33 @@ impl FuseNode {
         data: &FuseOpenPrivateData,
         lock_owner: u64,
     ) -> Result<(), SystemError> {
+        let writeback_cache = self.conn().has_init_flag(FUSE_WRITEBACK_CACHE);
+        // Linux skips both data-cache writeback and protocol FLUSH for
+        // FOPEN_NOFLUSH only when writeback-cache is disabled.
+        if !writeback_cache && (data.fopen_flags & FOPEN_NOFLUSH) != 0 {
+            return Ok(());
+        }
+
         // Linux fuse_flush() drains inode writeback even when this particular
         // descriptor is read-only: another writable open may have admitted
         // dirty data for the same inode.
-        let writeback_cache = self.conn().has_init_flag(FUSE_WRITEBACK_CACHE);
-        let _barrier = writeback_cache.then(|| self.writeback_barrier.write());
-        let writeback_result = if writeback_cache {
-            self.sync_dirty_cached_pages()
-        } else {
-            Ok(())
-        };
-        let wb_error_result = self.check_and_advance_open_wb_error(data);
-
+        let barrier = self.writeback_barrier.write();
+        // Linux fuse_flush() calls write_inode_now() in both cache modes. This
+        // matters when a shared mapping dirtied a cached page while the file
+        // descriptor is being closed, even without WRITEBACK_CACHE.
+        let writeback_result = self.sync_dirty_cached_pages_admitted(&barrier);
+        // Linux leaves NOWRITE after fuse_sync_writes(), before the blocking
+        // daemon FLUSH request. WRITE-before-FLUSH is already guaranteed by
+        // the synchronous drain, so retaining the barrier would only block new
+        // dirty admission and expose a notify/daemon reverse-wait cycle.
+        drop(barrier);
         writeback_result?;
-        wb_error_result?;
+        self.check_and_advance_open_wb_error(data)?;
 
-        if data.no_open || (data.fopen_flags & FOPEN_NOFLUSH) != 0 || self.conn().no_flush() {
+        // FOPEN_NOFLUSH only suppresses FLUSH when writeback-cache is disabled.
+        // With writeback-cache Linux must still send FLUSH after draining dirty
+        // pages so daemon-side flush/lock semantics and errors are preserved.
+        if self.conn().no_flush() {
             return Ok(());
         }
 
@@ -865,7 +1026,7 @@ impl FuseNode {
         Err(SystemError::EIO)
     }
 
-    fn writeback_page_with_handle(
+    fn writeback_pages_with_handle(
         &self,
         page_index: usize,
         buf: &[u8],
@@ -925,12 +1086,28 @@ impl FuseNode {
             if out.size as usize != chunk {
                 return Err(SystemError::EIO);
             }
+            self.check_not_stale()?;
+            let reply_generation = self.generation();
+            if handle.open_context.node_generation != 0
+                && reply_generation != 0
+                && handle.open_context.node_generation != reply_generation
+            {
+                return Err(SystemError::ESTALE);
+            }
             total += chunk;
         }
 
         Ok(total)
     }
 
+    fn writeback_page_with_handle(
+        &self,
+        page_index: usize,
+        buf: &[u8],
+    ) -> Result<usize, SystemError> {
+        self.writeback_pages_with_handle(page_index, buf)
+    }
+
     pub(super) fn set_open_private_data(
         &self,
         data: &mut FilePrivateData,
@@ -1044,7 +1221,7 @@ impl FuseNode {
         if atomic_dax_truncate {
             self.sync_dirty_cached_pages()?;
         }
-        let _dax_truncate_layout = if atomic_dax_truncate {
+        let dax_truncate_layout = if atomic_dax_truncate {
             Some(self.dax_layout_write_for_all()?)
         } else {
             None
@@ -1052,7 +1229,11 @@ impl FuseNode {
         if atomic_dax_truncate {
             // Close dirty admission between the optimistic drain and layout
             // exclusivity before OPEN can atomically change the host file.
-            self.sync_dirty_cached_pages()?;
+            self.sync_dirty_cached_pages_admitted(
+                dax_truncate_layout
+                    .as_ref()
+                    .expect("atomic DAX truncate guard"),
+            )?;
         }
         let file_flags = flags.bits();
         if self.conn.should_skip_open(opcode) {
@@ -1927,6 +2108,7 @@ impl FuseNode {
         len: usize,
         data: &FuseOpenPrivateData,
         discard_clean: bool,
+        _guard: &RwSemWriteGuard<'_, ()>,
     ) -> Result<(), SystemError> {
         let Some((start_page_index, end_page_index, end_page_exclusive)) =
             Self::direct_io_page_range(offset, len)?
@@ -1937,12 +2119,11 @@ impl FuseNode {
             return Ok(());
         };
 
-        page_cache
-            .manager()
-            .writeback_range(start_page_index, end_page_index)?;
-        page_cache
-            .manager()
-            .wait_writeback_range(start_page_index, end_page_index)?;
+        page_cache.manager().sync_range_with_stable_size(
+            start_page_index,
+            end_page_index,
+            self.stable_writeback_size()?,
+        )?;
         self.check_and_advance_open_wb_error(data)?;
 
         if discard_clean {

kernel/src/filesystem/fuse/inode/vfs.rs

@@ -43,7 +43,7 @@ impl IndexNode for FuseNode {
     }
 
     fn append_lock_fs(&self) -> Option<Arc<dyn FileSystem>> {
-        Some(self.fs())
+        self.try_fs()
     }
 
     fn as_any_ref(&self) -> &dyn core::any::Any {
@@ -400,7 +400,7 @@ impl IndexNode for FuseNode {
 
         if (fopen_flags & FOPEN_DIRECT_IO) != 0 || (file_flags & FileFlags::O_DIRECT.bits()) != 0 {
             let _barrier = self.writeback_barrier.write();
-            self.prepare_direct_io_range(offset, len, &private_data, false)?;
+            self.prepare_direct_io_range(offset, len, &private_data, false, &_barrier)?;
             let lock_owner = crate::filesystem::vfs::vcore::current_file_lock_owner_id();
             return self.read_direct_with_open(offset, len, buf, fh, file_flags, lock_owner);
         }
@@ -482,7 +482,15 @@ impl IndexNode for FuseNode {
         };
         let mut total_written = 0usize;
         if !cached_write {
-            self.prepare_direct_io_range(offset, len, &private_data, true)?;
+            self.prepare_direct_io_range(
+                offset,
+                len,
+                &private_data,
+                true,
+                _direct_write_guard
+                    .as_ref()
+                    .expect("direct writeback admission guard"),
+            )?;
         }
         if cached_write && self.conn().has_init_flag(FUSE_WRITEBACK_CACHE) {
             while total_written < len {
@@ -612,7 +620,9 @@ impl IndexNode for FuseNode {
             writeback_cache.then(|| self.writeback_barrier.write())
         };
         if writeback_cache {
-            self.sync_dirty_cached_pages()?;
+            self.sync_dirty_cached_pages_admitted(
+                _barrier.as_ref().expect("writeback admission guard"),
+            )?;
         }
         if metadata.size > old.size {
             self.resolve_pending_short_read_truncate(metadata.size.max(0) as usize)?;
@@ -793,7 +803,7 @@ impl IndexNode for FuseNode {
         };
         if changes_contents {
             // Close dirty admission between the first drain and exclusivity.
-            self.sync_dirty_cached_pages()?;
+            self.sync_dirty_cached_pages_admitted(&_barrier)?;
         }
 
         let in_arg = FuseFallocateIn {
@@ -822,7 +832,8 @@ impl IndexNode for FuseNode {
                     self.bump_attr_version();
                 }
                 drop(metadata);
-                self.cached_metadata_deadline_ns.store(0, Ordering::Relaxed);
+                self.cached_metadata_deadline_ticks
+                    .store(0, Ordering::Relaxed);
                 Ok(())
             }
             Err(SystemError::ENOSYS) => {
@@ -862,7 +873,7 @@ impl IndexNode for FuseNode {
         match fuse_data {
             FuseFilePrivateData::File(p) => {
                 let _barrier = self.writeback_barrier.write();
-                let sync_result = self.sync_cached_pages();
+                let sync_result = self.sync_dirty_cached_pages_admitted(&_barrier);
                 let wb_error_result = self.check_and_advance_open_wb_error(&p);
                 sync_result?;
                 wb_error_result?;
@@ -891,11 +902,9 @@ impl IndexNode for FuseNode {
 
         if let FuseFilePrivateData::File(_) = &fuse_data {
             let _barrier = self.writeback_barrier.write();
-            if let Some(page_cache) = self.cached_page_cache() {
-                let start_index = start >> MMArch::PAGE_SHIFT;
-                let end_index = end >> MMArch::PAGE_SHIFT;
-                page_cache.manager().sync_range(start_index, end_index)?;
-            }
+            let start_index = start >> MMArch::PAGE_SHIFT;
+            let end_index = end >> MMArch::PAGE_SHIFT;
+            self.sync_cached_range_admitted(start_index, end_index, &_barrier)?;
             if let FuseFilePrivateData::File(p) = fuse_data {
                 self.check_and_advance_open_wb_error(&p)?;
                 return self.fsync_with_fh(FUSE_FSYNC, p.fh, datasync);
@@ -914,6 +923,10 @@ impl IndexNode for FuseNode {
         self.fs.upgrade().unwrap()
     }
 
+    fn try_fs(&self) -> Option<Arc<dyn FileSystem>> {
+        self.fs.upgrade().map(|fs| fs as Arc<dyn FileSystem>)
+    }
+
     fn list(&self) -> Result<Vec<String>, SystemError> {
         self.check_not_stale()?;
         self.ensure_dir()?;

kernel/src/filesystem/fuse/stats.rs

@@ -46,10 +46,13 @@ static STATS_MODE: AtomicU8 = AtomicU8::new(FuseStatsMode::Off as u8);
 pub struct FuseStatsSnapshot {
     pub init_epoch: u64,
     pub negotiated_max_read_bytes: u64,
+    pub negotiated_max_write_bytes: u64,
     pub negotiated_max_pages: u64,
     pub negotiated_max_readahead_bytes: u64,
     pub negotiated_async_read: u64,
+    pub negotiated_writeback_cache: u64,
     pub effective_read_payload_limit_bytes: u64,
+    pub effective_write_payload_limit_bytes: u64,
     pub request_queue_current: u64,
     pub dispatch_current: u64,
     pub processing_current: u64,
@@ -79,8 +82,14 @@ pub struct FuseStatsSnapshot {
     pub readahead_saturated_single_page_extensions_total: u64,
     pub readahead_reservation_conflicts_total: u64,
     pub readahead_short_reads_total: u64,
+    pub write_requested_requests_total: u64,
+    pub write_requested_bytes_total: u64,
+    pub write_requested_bytes_max: u64,
     pub background_inflight_current: u64,
     pub read_reservation_current: u64,
+    pub invalidation_launder_batches_total: u64,
+    pub invalidation_launder_pages_total: u64,
+    pub mmap_writeback_admission_retries_total: u64,
     pub background_inflight_peak: u64,
     pub background_max_blocked_total: u64,
     pub background_congestion_skipped_total: u64,
@@ -228,10 +237,13 @@ static REQUESTS_QUEUED_TOTAL: AtomicU64 = AtomicU64::new(0);
 static INIT_EPOCH: AtomicU64 = AtomicU64::new(0);
 static INIT_LIMITS_SEQ: AtomicU64 = AtomicU64::new(0);
 static NEGOTIATED_MAX_READ_BYTES: AtomicU64 = AtomicU64::new(0);
+static NEGOTIATED_MAX_WRITE_BYTES: AtomicU64 = AtomicU64::new(0);
 static NEGOTIATED_MAX_PAGES: AtomicU64 = AtomicU64::new(0);
 static NEGOTIATED_MAX_READAHEAD_BYTES: AtomicU64 = AtomicU64::new(0);
 static NEGOTIATED_ASYNC_READ: AtomicU64 = AtomicU64::new(0);
+static NEGOTIATED_WRITEBACK_CACHE: AtomicU64 = AtomicU64::new(0);
 static EFFECTIVE_READ_PAYLOAD_LIMIT_BYTES: AtomicU64 = AtomicU64::new(0);
+static EFFECTIVE_WRITE_PAYLOAD_LIMIT_BYTES: AtomicU64 = AtomicU64::new(0);
 static REQUESTS_DEQUEUED_TOTAL: AtomicU64 = AtomicU64::new(0);
 static REQUEST_QUEUE_CURRENT: AtomicU64 = AtomicU64::new(0);
 static DISPATCH_CURRENT: AtomicU64 = AtomicU64::new(0);
@@ -264,14 +276,20 @@ static BACKGROUND_INFLIGHT_CURRENT: AtomicU64 = AtomicU64::new(0);
 static BACKGROUND_INFLIGHT_PEAK: AtomicU64 = AtomicU64::new(0);
 static BACKGROUND_MAX_BLOCKED_TOTAL: AtomicU64 = AtomicU64::new(0);
 static BACKGROUND_CONGESTION_SKIPPED_TOTAL: AtomicU64 = AtomicU64::new(0);
+static MMAP_WRITEBACK_ADMISSION_RETRIES_TOTAL: AtomicU64 = AtomicU64::new(0);
 
-pub fn on_fuse_read_limits_negotiated(
-    max_read: usize,
-    max_pages: usize,
-    max_readahead: usize,
-    async_read: bool,
-    effective_read_payload_limit: usize,
-) {
+pub struct NegotiatedFuseIoLimits {
+    pub max_read: usize,
+    pub max_write: usize,
+    pub max_pages: usize,
+    pub max_readahead: usize,
+    pub async_read: bool,
+    pub writeback_cache: bool,
+    pub effective_read_payload_limit: usize,
+    pub effective_write_payload_limit: usize,
+}
+
+pub fn on_fuse_io_limits_negotiated(limits: NegotiatedFuseIoLimits) {
     let sequence = loop {
         let current = INIT_LIMITS_SEQ.load(Ordering::Acquire);
         if current & 1 != 0 {
@@ -290,17 +308,25 @@ pub fn on_fuse_read_limits_negotiated(
             break current;
         }
     };
-    NEGOTIATED_MAX_READ_BYTES.store(max_read as u64, Ordering::Relaxed);
-    NEGOTIATED_MAX_PAGES.store(max_pages as u64, Ordering::Relaxed);
-    NEGOTIATED_MAX_READAHEAD_BYTES.store(max_readahead as u64, Ordering::Relaxed);
-    NEGOTIATED_ASYNC_READ.store(async_read as u64, Ordering::Relaxed);
-    EFFECTIVE_READ_PAYLOAD_LIMIT_BYTES
-        .store(effective_read_payload_limit as u64, Ordering::Relaxed);
+    NEGOTIATED_MAX_READ_BYTES.store(limits.max_read as u64, Ordering::Relaxed);
+    NEGOTIATED_MAX_WRITE_BYTES.store(limits.max_write as u64, Ordering::Relaxed);
+    NEGOTIATED_MAX_PAGES.store(limits.max_pages as u64, Ordering::Relaxed);
+    NEGOTIATED_MAX_READAHEAD_BYTES.store(limits.max_readahead as u64, Ordering::Relaxed);
+    NEGOTIATED_ASYNC_READ.store(limits.async_read as u64, Ordering::Relaxed);
+    NEGOTIATED_WRITEBACK_CACHE.store(limits.writeback_cache as u64, Ordering::Relaxed);
+    EFFECTIVE_READ_PAYLOAD_LIMIT_BYTES.store(
+        limits.effective_read_payload_limit as u64,
+        Ordering::Relaxed,
+    );
+    EFFECTIVE_WRITE_PAYLOAD_LIMIT_BYTES.store(
+        limits.effective_write_payload_limit as u64,
+        Ordering::Relaxed,
+    );
     INIT_EPOCH.fetch_add(1, Ordering::Relaxed);
     INIT_LIMITS_SEQ.store(sequence.wrapping_add(2), Ordering::Release);
 }
 
-fn fuse_init_limits_snapshot() -> (u64, u64, u64, u64, u64, u64) {
+fn fuse_init_limits_snapshot() -> (u64, u64, u64, u64, u64, u64, u64, u64, u64) {
     loop {
         let before = INIT_LIMITS_SEQ.load(Ordering::Acquire);
         if before & 1 != 0 {
@@ -310,10 +336,13 @@ fn fuse_init_limits_snapshot() -> (u64, u64, u64, u64, u64, u64) {
         let values = (
             INIT_EPOCH.load(Ordering::Relaxed),
             NEGOTIATED_MAX_READ_BYTES.load(Ordering::Relaxed),
+            NEGOTIATED_MAX_WRITE_BYTES.load(Ordering::Relaxed),
             NEGOTIATED_MAX_PAGES.load(Ordering::Relaxed),
             NEGOTIATED_MAX_READAHEAD_BYTES.load(Ordering::Relaxed),
             NEGOTIATED_ASYNC_READ.load(Ordering::Relaxed),
+            NEGOTIATED_WRITEBACK_CACHE.load(Ordering::Relaxed),
             EFFECTIVE_READ_PAYLOAD_LIMIT_BYTES.load(Ordering::Relaxed),
+            EFFECTIVE_WRITE_PAYLOAD_LIMIT_BYTES.load(Ordering::Relaxed),
         );
         if before == INIT_LIMITS_SEQ.load(Ordering::Acquire) {
             return values;
@@ -375,6 +404,14 @@ pub fn on_background_pressure(speculative: bool) {
     }
 }
 
+/// Count mmap write faults which must retry after dropping the address-space
+/// guard because an exclusive cached-writeback operation owns the inode gate.
+/// This is always-on: it is a correctness/lifecycle observation rather than
+/// optional hot-path detail, and the increment occurs only on contention.
+pub fn on_mmap_writeback_admission_retry() {
+    inc(&MMAP_WRITEBACK_ADMISSION_RETRIES_TOTAL);
+}
+
 static DEVICE_QUEUE_DEPTH_MAX: AtomicU64 = AtomicU64::new(0);
 static HIPRIO_VRING_SIZE_CONFIGURED: AtomicU64 = AtomicU64::new(0);
 static REQUEST_QUEUE_COUNT_CONFIGURED: AtomicU64 = AtomicU64::new(0);
@@ -440,6 +477,13 @@ static READ_REQUESTED_BYTES_MAX: AtomicU64 = AtomicU64::new(0);
 static READ_REQUESTED_PAGES: [AtomicU64; READ_PAGE_BUCKETS] =
     [const { AtomicU64::new(0) }; READ_PAGE_BUCKETS];
 
+// WRITE request-size diagnostics use FuseWriteIn.size at the successful
+// virtqueue submission point. They are optional so normal benchmarks do not
+// pay atomic RMW costs or mistake protocol/header bytes for file data.
+static WRITE_REQUESTED_REQUESTS_TOTAL: AtomicU64 = AtomicU64::new(0);
+static WRITE_REQUESTED_BYTES_TOTAL: AtomicU64 = AtomicU64::new(0);
+static WRITE_REQUESTED_BYTES_MAX: AtomicU64 = AtomicU64::new(0);
+
 static OPCODE_REQUESTS_TOTAL: [AtomicU64; OPCODE_BUCKETS] =
     [const { AtomicU64::new(0) }; OPCODE_BUCKETS];
 static OPCODE_REQUEST_BYTES_TOTAL: [AtomicU64; OPCODE_BUCKETS] =
@@ -1034,6 +1078,24 @@ pub fn on_virtiofs_read_requested(requested_bytes: usize, pages: usize) {
     inc(&READ_REQUESTED_PAGES[read_page_bucket(pages)]);
 }
 
+/// Record the file-data payload declared by a successfully submitted
+/// FUSE_WRITE. The caller obtains this value from `FuseWriteIn.size`.
+#[inline]
+pub fn on_virtiofs_write_requested(requested_bytes: usize) {
+    if !light_stats_enabled() || requested_bytes == 0 {
+        return;
+    }
+    inc(&WRITE_REQUESTED_REQUESTS_TOTAL);
+    add(&WRITE_REQUESTED_BYTES_TOTAL, requested_bytes as u64);
+    update_peak(&WRITE_REQUESTED_BYTES_MAX, requested_bytes as u64);
+}
+
+/// Whether callers should pay the cost of decoding WRITE request details.
+#[inline]
+pub fn write_request_stats_enabled() -> bool {
+    light_stats_enabled()
+}
+
 /// Queue acceptance is the DMA ownership commit point.
 #[inline]
 pub fn on_virtiofs_direct_read_requested(requested_bytes: usize) {
@@ -1141,15 +1203,28 @@ pub fn on_virtiofs_dax_device_reset() {
 }
 
 pub fn fuse_snapshot() -> FuseStatsSnapshot {
-    let (init_epoch, max_read, max_pages, max_readahead, async_read, effective_read) =
-        fuse_init_limits_snapshot();
+    let (
+        init_epoch,
+        max_read,
+        max_write,
+        max_pages,
+        max_readahead,
+        async_read,
+        writeback_cache,
+        effective_read,
+        effective_write,
+    ) = fuse_init_limits_snapshot();
+    let page_cache = page_cache_stats::snapshot();
     FuseStatsSnapshot {
         init_epoch,
         negotiated_max_read_bytes: max_read,
+        negotiated_max_write_bytes: max_write,
         negotiated_max_pages: max_pages,
         negotiated_max_readahead_bytes: max_readahead,
         negotiated_async_read: async_read,
+        negotiated_writeback_cache: writeback_cache,
         effective_read_payload_limit_bytes: effective_read,
+        effective_write_payload_limit_bytes: effective_write,
         request_queue_current: REQUEST_QUEUE_CURRENT.load(Ordering::Acquire),
         dispatch_current: DISPATCH_CURRENT.load(Ordering::Acquire),
         processing_current: PROCESSING_CURRENT.load(Ordering::Acquire),
@@ -1185,8 +1260,15 @@ pub fn fuse_snapshot() -> FuseStatsSnapshot {
         readahead_reservation_conflicts_total: READAHEAD_RESERVATION_CONFLICTS_TOTAL
             .load(Ordering::Relaxed),
         readahead_short_reads_total: READAHEAD_SHORT_READS_TOTAL.load(Ordering::Relaxed),
+        write_requested_requests_total: WRITE_REQUESTED_REQUESTS_TOTAL.load(Ordering::Relaxed),
+        write_requested_bytes_total: WRITE_REQUESTED_BYTES_TOTAL.load(Ordering::Relaxed),
+        write_requested_bytes_max: WRITE_REQUESTED_BYTES_MAX.load(Ordering::Relaxed),
         background_inflight_current: BACKGROUND_INFLIGHT_CURRENT.load(Ordering::Acquire),
-        read_reservation_current: page_cache_stats::snapshot().read_dma_reservations,
+        read_reservation_current: page_cache.read_dma_reservations,
+        invalidation_launder_batches_total: page_cache.invalidation_launder_batches,
+        invalidation_launder_pages_total: page_cache.invalidation_launder_pages,
+        mmap_writeback_admission_retries_total: MMAP_WRITEBACK_ADMISSION_RETRIES_TOTAL
+            .load(Ordering::Relaxed),
         background_inflight_peak: BACKGROUND_INFLIGHT_PEAK.load(Ordering::Relaxed),
         background_max_blocked_total: BACKGROUND_MAX_BLOCKED_TOTAL.load(Ordering::Relaxed),
         background_congestion_skipped_total: BACKGROUND_CONGESTION_SKIPPED_TOTAL
@@ -1302,10 +1384,13 @@ detailed opcode,copy,allocation\n\
 [fuse]\n\
 init_epoch {}\n\
 negotiated_max_read_bytes {}\n\
+negotiated_max_write_bytes {}\n\
 negotiated_max_pages {}\n\
 negotiated_max_readahead_bytes {}\n\
 negotiated_async_read {}\n\
+negotiated_writeback_cache {}\n\
 effective_read_payload_limit_bytes {}\n\
+effective_write_payload_limit_bytes {}\n\
 request_queue_current {}\n\
 dispatch_current {}\n\
 processing_current {}\n\
@@ -1335,8 +1420,14 @@ readahead_window_extension_pages_total {}\n\
 readahead_saturated_single_page_extensions_total {}\n\
 readahead_reservation_conflicts_total {}\n\
 readahead_short_reads_total {}\n\
+write_requested_requests_total {}\n\
+write_requested_bytes_total {}\n\
+write_requested_bytes_max {}\n\
 background_inflight_current {}\n\
 read_reservation_current {}\n\
+invalidation_launder_batches_total {}\n\
+invalidation_launder_pages_total {}\n\
+mmap_writeback_admission_retries_total {}\n\
 background_inflight_peak {}\n\
 background_max_blocked_total {}\n\
 background_congestion_skipped_total {}\n\
@@ -1434,10 +1525,13 @@ dax_device_resets_total {}\n",
         stats_mode().as_str(),
         fuse.init_epoch,
         fuse.negotiated_max_read_bytes,
+        fuse.negotiated_max_write_bytes,
         fuse.negotiated_max_pages,
         fuse.negotiated_max_readahead_bytes,
         fuse.negotiated_async_read,
+        fuse.negotiated_writeback_cache,
         fuse.effective_read_payload_limit_bytes,
+        fuse.effective_write_payload_limit_bytes,
         fuse.request_queue_current,
         fuse.dispatch_current,
         fuse.processing_current,
@@ -1467,8 +1561,14 @@ dax_device_resets_total {}\n",
         fuse.readahead_saturated_single_page_extensions_total,
         fuse.readahead_reservation_conflicts_total,
         fuse.readahead_short_reads_total,
+        fuse.write_requested_requests_total,
+        fuse.write_requested_bytes_total,
+        fuse.write_requested_bytes_max,
         fuse.background_inflight_current,
         fuse.read_reservation_current,
+        fuse.invalidation_launder_batches_total,
+        fuse.invalidation_launder_pages_total,
+        fuse.mmap_writeback_admission_retries_total,
         fuse.background_inflight_peak,
         fuse.background_max_blocked_total,
         fuse.background_congestion_skipped_total,

kernel/src/filesystem/fuse/virtiofs/bridge.rs

@@ -30,8 +30,8 @@ use crate::{
 use super::super::{
     conn::{FuseConn, FuseReplyCapacitySource, FuseReplyContract, FuseRequest},
     protocol::{
-        fuse_pack_struct, fuse_read_struct, FuseOutHeader, FUSE_DESTROY, FUSE_FORGET,
-        FUSE_INTERRUPT, FUSE_READ,
+        fuse_pack_struct, fuse_read_struct, FuseInHeader, FuseOutHeader, FuseWriteIn, FUSE_DESTROY,
+        FUSE_FORGET, FUSE_INTERRUPT, FUSE_READ, FUSE_WRITE,
     },
     stats, trace,
 };
@@ -906,6 +906,17 @@ impl VirtioFsBridgeContext {
         } else {
             None
         };
+        let write_requested_bytes =
+            if trace_opcode == FUSE_WRITE && stats::write_request_stats_enabled() {
+                pending
+                    .req
+                    .bytes()
+                    .get(core::mem::size_of::<FuseInHeader>()..)
+                    .and_then(|payload| fuse_read_struct::<FuseWriteIn>(payload).ok())
+                    .map(|input| input.size as usize)
+            } else {
+                None
+            };
         let queue_size = match kind {
             QueueKind::Hiprio => self.hiprio_vq.as_ref().ok_or(SystemError::EIO)?.size(),
             QueueKind::Request(slot) => self
@@ -1083,6 +1094,9 @@ impl VirtioFsBridgeContext {
         };
         debug_assert!(replaced.is_none());
         stats::on_virtiofs_submitted(trace_opcode, req_len);
+        if let Some(requested_bytes) = write_requested_bytes {
+            stats::on_virtiofs_write_requested(requested_bytes);
+        }
         if let Some(requested_bytes) = read_requested_bytes {
             stats::on_virtiofs_read_requested(
                 requested_bytes,

kernel/src/filesystem/vfs/mod.rs

@@ -976,6 +976,14 @@ pub trait IndexNode: Any + Sync + Send + Debug + CastFromSync {
     /// @brief 获取inode所在的文件系统的指针
     fn fs(&self) -> Arc<dyn FileSystem>;
 
+    /// 获取 inode 所在的文件系统；供异步回收等可能与卸载并发的路径使用。
+    ///
+    /// 默认实现适用于 inode 强持有文件系统的实现。仅当 inode 与文件系统之间
+    /// 使用弱引用、且调用方允许文件系统已完成销毁时才应覆盖此方法。
+    fn try_fs(&self) -> Option<Arc<dyn FileSystem>> {
+        Some(self.fs())
+    }
+
     /// @brief 获取当前 inode 所在挂载点的挂载标志
     fn mount_flags(&self) -> MountFlags {
         MountFlags::empty()

kernel/src/filesystem/vfs/syscall/sys_sync_file_range.rs

@@ -113,7 +113,9 @@ fn sync_file_range(
         file.check_and_advance_wb_error(&page_cache)?;
     }
     if flags.contains(SyncFileRangeFlags::WRITE) {
-        manager.start_writeback_range(start_index, end_index)?;
+        let sync_all =
+            flags.contains(SyncFileRangeFlags::WAIT_BEFORE | SyncFileRangeFlags::WAIT_AFTER);
+        manager.start_writeback_range(start_index, end_index, sync_all)?;
     }
     if flags.contains(SyncFileRangeFlags::WAIT_AFTER) {
         manager.wait_writeback_range(start_index, end_index)?;

kernel/src/mm/fault.rs

@@ -337,6 +337,11 @@ impl<'a> PageFaultMessage<'a> {
 /// 缺页中断处理结构体
 pub struct PageFaultHandler;
 
+enum FilemapMkwriteSize {
+    FetchFromInode,
+    Stable(usize),
+}
+
 impl PageFaultHandler {
     #[inline(always)]
     fn account_new_present_mapping(mm: &Arc<AddressSpace>) {
@@ -1185,6 +1190,24 @@ impl PageFaultHandler {
     }
 
     pub unsafe fn filemap_page_mkwrite(pfm: &mut PageFaultMessage) -> VmFaultReason {
+        Self::filemap_page_mkwrite_inner(pfm, FilemapMkwriteSize::FetchFromInode)
+    }
+
+    /// Prepare a shared writable file-backed page using an inode size which
+    /// the filesystem has already stabilized against truncate and dirty-page
+    /// admission.  Filesystems whose metadata operation can enter userspace
+    /// must use this entry point from a fault critical section.
+    pub unsafe fn filemap_page_mkwrite_with_stable_size(
+        pfm: &mut PageFaultMessage,
+        stable_size: usize,
+    ) -> VmFaultReason {
+        Self::filemap_page_mkwrite_inner(pfm, FilemapMkwriteSize::Stable(stable_size))
+    }
+
+    unsafe fn filemap_page_mkwrite_inner(
+        pfm: &mut PageFaultMessage,
+        size_source: FilemapMkwriteSize,
+    ) -> VmFaultReason {
         let vma = pfm.vma();
         let vma_guard = vma.lock();
         let file = vma_guard.vm_file().expect("no vm_file in vma");
@@ -1210,8 +1233,15 @@ impl PageFaultHandler {
             return VmFaultReason::VM_FAULT_RETRY;
         }
 
-        if let Ok(md) = file.inode().metadata() {
-            let size = md.size.max(0) as usize;
+        let size = match size_source {
+            FilemapMkwriteSize::FetchFromInode => file
+                .inode()
+                .metadata()
+                .ok()
+                .map(|metadata| metadata.size.max(0) as usize),
+            FilemapMkwriteSize::Stable(size) => Some(size),
+        };
+        if let Some(size) = size {
             if size == 0 || backing_pgoff.saturating_mul(MMArch::PAGE_SIZE) >= size {
                 return VmFaultReason::VM_FAULT_SIGBUS;
             }
@@ -1219,10 +1249,17 @@ impl PageFaultHandler {
 
         match page_cache.manager().prepare_page_mkwrite(page_index, &page) {
             Ok(()) => {}
-            Err(SystemError::EAGAIN_OR_EWOULDBLOCK) => return VmFaultReason::VM_FAULT_RETRY,
+            Err(SystemError::EAGAIN_OR_EWOULDBLOCK) => {
+                if let Some(wait) = page_cache
+                    .manager()
+                    .page_mkwrite_retry_wait(page_index, &page)
+                {
+                    pfm.set_retry_wait(wait);
+                }
+                return VmFaultReason::VM_FAULT_RETRY;
+            }
             Err(_) => return VmFaultReason::VM_FAULT_SIGBUS,
         }
-
         VmFaultReason::empty()
     }
 

kernel/src/mm/page.rs

@@ -17,7 +17,10 @@ use lru::LruCache;
 use crate::{
     arch::{mm::LockedFrameAllocator, MMArch},
     filesystem::{
-        page_cache::{list_page_caches, PageCache},
+        page_cache::{
+            async_writeback_progress_snapshot, list_page_caches, wait_for_async_writeback_progress,
+            PageCache,
+        },
         vfs::FilePrivateData,
     },
     init::initcall::INITCALL_CORE,
@@ -283,9 +286,22 @@ fn page_reclaim_thread() -> i32 {
         // 保留4096个页面，总计16MB的空闲空间
         if usage.free().data() < 4096 {
             let page_to_free = 4096;
+            let writeback_generation = async_writeback_progress_snapshot();
             // 分离选择和回收阶段，避免长时间持有页面回收器锁导致与
             // page_manager/page_cache 的锁顺序反转。
-            PageReclaimer::shrink_list(PageFrameCount::new(page_to_free));
+            let progress = PageReclaimer::shrink_list(PageFrameCount::new(page_to_free));
+            if progress.reclaimed == 0 {
+                // Any no-progress pass must yield: dirty pages may need an
+                // asynchronous writeback completion, while mapped,
+                // unevictable, or otherwise busy pages cannot be reclaimed by
+                // immediately draining the same LRU again.  If there is no
+                // writeback to wait for, use a bounded retry backoff.
+                if progress.dirty_seen == 0
+                    || !wait_for_async_writeback_progress(writeback_generation)
+                {
+                    let _ = nanosleep(PosixTimeSpec::new(0, 1_000_000));
+                }
+            }
         } else {
             //TODO Temporarily let page reclaim thread handle dirty page writeback; should be separated later.
             PageReclaimer::flush_dirty_pages();
@@ -306,12 +322,21 @@ pub fn page_reclaimer_lock() -> MutexGuard<'static, PageReclaimer> {
 /// 页面回收器
 pub struct PageReclaimer {
     lru: LruCache<PhysAddr, Arc<Page>>,
+    writeback_cursor: usize,
+}
+
+#[derive(Clone, Copy, Debug, Default)]
+pub struct ReclaimProgress {
+    pub reclaimed: usize,
+    pub dirty_seen: usize,
+    pub writeback_scheduled: usize,
 }
 
 impl PageReclaimer {
     pub fn new() -> Self {
         Self {
             lru: LruCache::unbounded(),
+            writeback_cursor: 0,
         }
     }
 
@@ -331,7 +356,7 @@ impl PageReclaimer {
     /// 分两阶段：
     /// 1) 持有回收器锁，从 LRU 中摘下目标页面列表；
     /// 2) 释放回收器锁后，逐个执行写回与回收，避免锁顺序反转。
-    pub fn shrink_list(count: PageFrameCount) {
+    pub fn shrink_list(count: PageFrameCount) -> ReclaimProgress {
         // 阶段1：仅持有回收器锁，摘取受害者
         let victims = {
             let mut reclaimer = page_reclaimer_lock();
@@ -339,7 +364,7 @@ impl PageReclaimer {
         };
 
         // 阶段2：不持有回收器锁，安全地回收页面
-        Self::evict_pages(victims);
+        Self::evict_pages(victims)
     }
 
     /// 从 LRU 中批量弹出页面，不做任何回收操作。
@@ -355,9 +380,10 @@ impl PageReclaimer {
     }
 
     /// 在不持有回收器锁的情况下，完成页面写回与回收。
-    fn evict_pages(victims: Vec<Arc<Page>>) {
+    fn evict_pages(victims: Vec<Arc<Page>>) -> ReclaimProgress {
+        let mut progress = ReclaimProgress::default();
         for page in victims {
-            let mut guard = page.write();
+            let guard = page.write();
             if let PageType::File(info) = guard.page_type().clone() {
                 if guard.flags().contains(PageFlags::PG_UNEVICTABLE) {
                     continue;
@@ -376,6 +402,8 @@ impl PageReclaimer {
                 let paddr = guard.phys_address();
 
                 if guard.flags().contains(PageFlags::PG_DIRTY) {
+                    progress.dirty_seen += 1;
+                    const MAX_RECLAIMER_PAGES_PER_BATCH: usize = 64;
                     let writeback_target = match guard.page_type() {
                         PageType::File(info) => info
                             .page_cache
@@ -386,22 +414,28 @@ impl PageReclaimer {
                     drop(guard);
 
                     if let Some((page_cache, page_index)) = writeback_target {
-                        let _ = page_cache.manager().writeback_page(page_index);
+                        // Reclaim must never synchronously enter a filesystem
+                        // or FUSE daemon. Schedule the same bounded, try-only
+                        // worker used by background writeback and keep the
+                        // victim on the LRU until a later pass observes it
+                        // clean. This also lets adjacent victims coalesce.
+                        let end = page_index.saturating_add(MAX_RECLAIMER_PAGES_PER_BATCH - 1);
+                        if page_cache
+                            .manager()
+                            .try_start_reclaimer_writeback_range(page_index, end)
+                            .unwrap_or(false)
+                        {
+                            progress.writeback_scheduled += 1;
+                        }
+                        page_reclaimer_lock().insert_page(paddr, &page);
+                        continue;
                     } else {
                         let mut guard = page.write();
                         guard.remove_flags(PageFlags::PG_DIRTY | PageFlags::PG_WRITEBACK);
                         drop(guard);
-                        page_manager_lock().remove_page(&paddr);
-                        continue;
-                    }
-
-                    guard = page.write();
-                    if guard.flags().intersects(
-                        PageFlags::PG_DIRTY | PageFlags::PG_WRITEBACK | PageFlags::PG_UNEVICTABLE,
-                    ) || guard.map_count() != 0
-                    {
-                        drop(guard);
-                        page_reclaimer_lock().insert_page(paddr, &page);
+                        if page_manager_lock().remove_page(&paddr).is_some() {
+                            progress.reclaimed += 1;
+                        }
                         continue;
                     }
                 }
@@ -427,9 +461,12 @@ impl PageReclaimer {
                         continue;
                     }
                 }
-                page_manager_lock().remove_page(&paddr);
+                if page_manager_lock().remove_page(&paddr).is_some() {
+                    progress.reclaimed += 1;
+                }
             }
         }
+        progress
     }
 
     /// Drop clean pagecache pages only, matching Linux drop_caches semantics.
@@ -542,32 +579,64 @@ impl PageReclaimer {
         }
     }
 
-    /// lru脏页刷新
-    fn dirty_pages_snapshot(&self) -> Vec<Arc<Page>> {
-        self.lru
-            .iter()
-            .filter_map(|(_paddr, page)| {
-                let guard = page.read();
-                if guard.flags().contains(PageFlags::PG_DIRTY) {
-                    Some(page.clone())
-                } else {
-                    None
-                }
-            })
-            .collect()
+    /// Take a bounded, rotating LRU snapshot without acquiring any page lock
+    /// while the global reclaimer lock is held.
+    fn writeback_scan_snapshot(&mut self) -> Vec<Arc<Page>> {
+        const MAX_RECLAIMER_SCAN_PAGES: usize = 512;
+        const RECLAIMER_SCAN_STRIDE: usize = 8;
+        let total = self.lru.len();
+        if total == 0 {
+            self.writeback_cursor = 0;
+            return Vec::new();
+        }
+
+        let start = self.writeback_cursor % total;
+        let count = total.min(MAX_RECLAIMER_SCAN_PAGES);
+        let mut pages = Vec::new();
+        if pages.try_reserve_exact(count).is_err() {
+            return pages;
+        }
+        pages.extend(
+            self.lru
+                .iter()
+                .skip(start)
+                .chain(self.lru.iter().take(start))
+                .take(count)
+                .map(|(_paddr, page)| page.clone()),
+        );
+        // Large LRUs advance by the whole sampled window so every region is
+        // visited before windows overlap. When the entire LRU fits, retain a
+        // small stride to rotate which hot cache gets the first scheduling
+        // slots without skipping any page from the snapshot.
+        let advance = if total > count {
+            count
+        } else {
+            RECLAIMER_SCAN_STRIDE.min(total)
+        };
+        self.writeback_cursor = if advance >= total - start {
+            advance - (total - start)
+        } else {
+            start + advance
+        };
+        pages
     }
 
     pub fn flush_dirty_pages() {
         let pages = {
-            let reclaimer = page_reclaimer_lock();
-            reclaimer.dirty_pages_snapshot()
+            let mut reclaimer = page_reclaimer_lock();
+            reclaimer.writeback_scan_snapshot()
         };
         Self::flush_dirty_pages_snapshot(pages);
     }
 
     fn flush_dirty_pages_snapshot(pages: Vec<Arc<Page>>) {
+        const MAX_RECLAIMER_RUNNERS: usize = 8;
+        const MAX_RECLAIMER_PAGES_PER_RUN: usize = 512;
+        let mut scheduled_runners = 0;
         for page in pages {
-            let mut guard = page.write();
+            let Some(guard) = page.try_read() else {
+                continue;
+            };
             if guard.flags().contains(PageFlags::PG_DIRTY) {
                 let writeback_target = match guard.page_type() {
                     PageType::File(info) => info
@@ -578,9 +647,28 @@ impl PageReclaimer {
                 };
                 if let Some((page_cache, page_index)) = writeback_target {
                     drop(guard);
-                    let _ = page_cache.manager().writeback_page(page_index);
+                    if scheduled_runners < MAX_RECLAIMER_RUNNERS {
+                        // Anchor every attempt at a page from the LRU snapshot,
+                        // then let one cache runner drain the same bounded
+                        // 512-page window as multiple backend-sized batches.
+                        // A contended cache is skipped immediately and later
+                        // snapshot candidates are still considered.
+                        let end = page_index.saturating_add(MAX_RECLAIMER_PAGES_PER_RUN - 1);
+                        if page_cache
+                            .manager()
+                            .try_start_reclaimer_writeback_range(page_index, end)
+                            .unwrap_or(false)
+                        {
+                            scheduled_runners += 1;
+                        }
+                    }
                 } else {
-                    Self::page_writeback(&mut guard, false);
+                    drop(guard);
+                    if let Some(mut guard) = page.try_write() {
+                        if guard.flags().contains(PageFlags::PG_DIRTY) {
+                            Self::page_writeback(&mut guard, false);
+                        }
+                    }
                 }
             }
         }
@@ -682,6 +770,10 @@ impl Page {
         self.inner.read()
     }
 
+    pub fn try_read(&self) -> Option<RwSemReadGuard<'_, InnerPage>> {
+        self.inner.try_read()
+    }
+
     pub fn upread(&self) -> RwSemUpgradeableGuard<'_, InnerPage> {
         self.inner.upread()
     }
@@ -689,6 +781,10 @@ impl Page {
     pub fn write(&self) -> RwSemWriteGuard<'_, InnerPage> {
         self.inner.write()
     }
+
+    pub fn try_write(&self) -> Option<RwSemWriteGuard<'_, InnerPage>> {
+        self.inner.try_write()
+    }
 }
 
 #[derive(Debug)]

kernel/src/mm/page_cache_stats.rs

@@ -10,6 +10,8 @@ pub struct PageCacheStatsSnapshot {
     pub unevictable: u64,
     pub drop_pagecache: u64,
     pub read_dma_reservations: u64,
+    pub invalidation_launder_batches: u64,
+    pub invalidation_launder_pages: u64,
 }
 
 static FILE_PAGES: AtomicU64 = AtomicU64::new(0);
@@ -20,6 +22,8 @@ static SHMEM_PAGES: AtomicU64 = AtomicU64::new(0);
 static UNEVICTABLE: AtomicU64 = AtomicU64::new(0);
 static DROP_PAGECACHE: AtomicU64 = AtomicU64::new(0);
 static READ_DMA_RESERVATIONS: AtomicU64 = AtomicU64::new(0);
+static INVALIDATION_LAUNDER_BATCHES: AtomicU64 = AtomicU64::new(0);
+static INVALIDATION_LAUNDER_PAGES: AtomicU64 = AtomicU64::new(0);
 
 #[inline]
 pub fn inc_file_pages() {
@@ -96,6 +100,12 @@ pub fn end_read_dma_reservation() {
     READ_DMA_RESERVATIONS.fetch_sub(1, Ordering::AcqRel);
 }
 
+#[inline]
+pub fn record_invalidation_launder_batch(pages: usize) {
+    INVALIDATION_LAUNDER_BATCHES.fetch_add(1, Ordering::Relaxed);
+    INVALIDATION_LAUNDER_PAGES.fetch_add(pages as u64, Ordering::Relaxed);
+}
+
 #[inline]
 pub fn snapshot() -> PageCacheStatsSnapshot {
     PageCacheStatsSnapshot {
@@ -107,5 +117,7 @@ pub fn snapshot() -> PageCacheStatsSnapshot {
         unevictable: UNEVICTABLE.load(Ordering::Relaxed),
         drop_pagecache: DROP_PAGECACHE.load(Ordering::Relaxed),
         read_dma_reservations: READ_DMA_RESERVATIONS.load(Ordering::Acquire),
+        invalidation_launder_batches: INVALIDATION_LAUNDER_BATCHES.load(Ordering::Relaxed),
+        invalidation_launder_pages: INVALIDATION_LAUNDER_PAGES.load(Ordering::Relaxed),
     }
 }

kernel/src/process/kthread.rs

@@ -1,6 +1,6 @@
 use core::{
     hint::spin_loop,
-    sync::atomic::{compiler_fence, AtomicBool, Ordering},
+    sync::atomic::{AtomicBool, AtomicUsize, Ordering},
 };
 
 use alloc::{
@@ -17,7 +17,7 @@ use crate::{
     arch::CurrentIrqArch,
     exception::{irqdesc::IrqAction, InterruptArch},
     init::initial_kthread::{initial_kernel_thread, set_system_state, SystemState},
-    libs::{cpumask::CpuMask, once::Once, spinlock::SpinLock},
+    libs::{once::Once, spinlock::SpinLock, wait_queue::WaitQueue},
     process::{ProcessManager, ProcessState},
     sched::{completion::Completion, schedule, SchedMode},
     smp::cpu::ProcessorId,
@@ -29,7 +29,21 @@ use super::{fork::CloneFlags, ProcessControlBlock, ProcessFlags, RawPid};
 static KTHREAD_CREATE_LIST: SpinLock<LinkedList<Arc<KernelThreadCreateInfo>>> =
     SpinLock::new(LinkedList::new());
 
-static mut KTHREAD_DAEMON_PCB: Option<Arc<ProcessControlBlock>> = None;
+/// All work that can make kthreadd useful must use this notification path.
+/// The pending bit coalesces events because each daemon pass drains the full
+/// create list and reaps every zombie child.
+static KTHREAD_DAEMON_WAIT: WaitQueue = WaitQueue::default();
+static KTHREAD_DAEMON_WORK_PENDING: AtomicBool = AtomicBool::new(false);
+static KTHREAD_DAEMON_READY: AtomicBool = AtomicBool::new(false);
+static KTHREAD_SELFTEST_RUNNING: AtomicBool = AtomicBool::new(false);
+
+struct KthreadSelftestGuard;
+
+impl Drop for KthreadSelftestGuard {
+    fn drop(&mut self) {
+        KTHREAD_SELFTEST_RUNNING.store(false, Ordering::Release);
+    }
+}
 
 #[derive(Debug)]
 pub enum WorkerPrivate {
@@ -146,8 +160,9 @@ pub struct KernelThreadCreateInfo {
     closure: SpinLock<Option<Box<KernelThreadClosure>>>,
     /// 内核线程的名字
     name: String,
-    /// 是否已经完成创建 todo:使用comletion机制优化这里
+    /// 是否已经完成创建
     created: AtomicKernelThreadCreateStatus,
+    created_completion: Completion,
     result_pcb: SpinLock<Option<Arc<ProcessControlBlock>>>,
     /// 不安全的Arc引用计数，当内核线程创建失败时，需要减少这个计数
     has_unsafe_arc_instance: AtomicBool,
@@ -173,6 +188,7 @@ impl KernelThreadCreateInfo {
             closure: SpinLock::new(Some(Box::new(func))),
             name,
             created: AtomicKernelThreadCreateStatus::new(KernelThreadCreateStatus::NotCreated),
+            created_completion: Completion::new(),
             result_pcb: SpinLock::new(None),
             has_unsafe_arc_instance: AtomicBool::new(false),
             self_ref: Weak::new(),
@@ -197,23 +213,23 @@ impl KernelThreadCreateInfo {
     /// - Some(Arc<ProcessControlBlock>) 创建成功，返回新创建的内核线程的PCB
     /// - None 创建失败
     pub fn poll_result(&self) -> Option<Arc<ProcessControlBlock>> {
-        loop {
-            match self.created.load(Ordering::SeqCst) {
-                KernelThreadCreateStatus::Created => {
-                    return self.result_pcb.lock().take();
-                }
-                KernelThreadCreateStatus::NotCreated => {
-                    spin_loop();
-                }
-                KernelThreadCreateStatus::ErrorOccured => {
-                    // 创建失败，减少不安全的Arc引用计数
-                    let to_delete = self.has_unsafe_arc_instance.swap(false, Ordering::SeqCst);
-                    if to_delete {
-                        let self_ref = self.self_ref.upgrade().unwrap();
-                        unsafe { Arc::decrement_strong_count(&self_ref) };
-                    }
-                    return None;
+        self.created_completion
+            .wait_for_completion()
+            .expect("kthread create completion wait failed");
+
+        match self.created.load(Ordering::SeqCst) {
+            KernelThreadCreateStatus::Created => self.result_pcb.lock().take(),
+            KernelThreadCreateStatus::ErrorOccured => {
+                // 创建失败，减少不安全的Arc引用计数
+                let to_delete = self.has_unsafe_arc_instance.swap(false, Ordering::SeqCst);
+                if to_delete {
+                    let self_ref = self.self_ref.upgrade().unwrap();
+                    unsafe { Arc::decrement_strong_count(&self_ref) };
                 }
+                None
+            }
+            KernelThreadCreateStatus::NotCreated => {
+                panic!("kthread create completion published without a result")
             }
         }
     }
@@ -227,10 +243,16 @@ impl KernelThreadCreateInfo {
     }
 
     pub unsafe fn set_create_ok(&self, pcb: Arc<ProcessControlBlock>) {
-        // todo: 使用completion机制优化这里
         self.result_pcb.lock().replace(pcb);
         self.created
             .store(KernelThreadCreateStatus::Created, Ordering::SeqCst);
+        self.created_completion.complete();
+    }
+
+    pub fn set_create_error(&self) {
+        self.created
+            .store(KernelThreadCreateStatus::ErrorOccured, Ordering::SeqCst);
+        self.created_completion.complete();
     }
 
     /// 生成一个不安全的Arc指针（用于创建内核线程时传递参数）
@@ -308,7 +330,11 @@ impl KernelThreadCreateInfo {
         if flags.contains(KernelThreadFlags::IS_PER_CPU) {
             let cpu = (*self.bound_cpu.lock())
                 .expect("kthread create: per-cpu thread missing target cpu");
-            pcb.sched_info().set_cpus_allowed(CpuMask::from_cpu(cpu));
+            let allowed = pcb.sched_info().cpus_allowed();
+            assert!(
+                allowed.get(cpu).unwrap_or(false) && allowed.iter_cpu().count() == 1,
+                "kthread create: per-cpu affinity was not installed before first run"
+            );
         }
     }
 
@@ -374,16 +400,12 @@ impl KernelThreadMechanism {
             let info = KernelThreadCreateInfo::new(closure, "kthreadd".to_string());
             info.set_to_mark_sleep(false)
                 .expect("kthreadadd should be run first");
-            let kthreadd_pid: RawPid = Self::__inner_create(
+            Self::__inner_create(
                 &info,
                 CloneFlags::CLONE_VM | CloneFlags::CLONE_FS | CloneFlags::CLONE_FILES,
             )
             .expect("Failed to create kthread daemon");
-            let pcb = ProcessManager::find_task_by_vpid(kthreadd_pid).unwrap();
-            ProcessManager::wakeup(&pcb).expect("Failed to wakeup kthread daemon");
-            unsafe {
-                KTHREAD_DAEMON_PCB.replace(pcb);
-            }
+            KTHREAD_DAEMON_READY.store(true, Ordering::Release);
             info!("Initialize kernel thread mechanism stage2 complete");
         });
     }
@@ -401,17 +423,21 @@ impl KernelThreadMechanism {
     #[allow(dead_code)]
     pub fn create(func: KernelThreadClosure, name: String) -> Option<Arc<ProcessControlBlock>> {
         let info = KernelThreadCreateInfo::new(func, name);
-        while unsafe { KTHREAD_DAEMON_PCB.is_none() } {
+        while !KTHREAD_DAEMON_READY.load(Ordering::Acquire) {
             // 等待kthreadd启动
             spin_loop()
         }
         KTHREAD_CREATE_LIST.lock().push_back(info.clone());
-        compiler_fence(Ordering::SeqCst);
-        ProcessManager::wakeup(unsafe { KTHREAD_DAEMON_PCB.as_ref().unwrap() })
-            .expect("Failed to wakeup kthread daemon");
+        Self::notify_daemon();
         return info.poll_result();
     }
 
+    /// Notify kthreadd after publishing create or zombie-reap work.
+    pub(crate) fn notify_daemon() {
+        KTHREAD_DAEMON_WORK_PENDING.store(true, Ordering::Release);
+        KTHREAD_DAEMON_WAIT.wakeup(None);
+    }
+
     /// 创建并运行一个新的内核线程
     ///
     /// ## 参数
@@ -557,9 +583,17 @@ impl KernelThreadMechanism {
         drop(current_pcb);
 
         loop {
-            let mut list = KTHREAD_CREATE_LIST.lock();
-            while let Some(info) = list.pop_front() {
-                drop(list);
+            KTHREAD_DAEMON_WAIT
+                .wait_event_uninterruptible(
+                    || KTHREAD_DAEMON_WORK_PENDING.swap(false, Ordering::AcqRel),
+                    None::<fn()>,
+                )
+                .expect("kthreadd wait failed");
+
+            loop {
+                let Some(info) = KTHREAD_CREATE_LIST.lock().pop_front() else {
+                    break;
+                };
                 // create a new kernel thread
                 let result: Result<RawPid, SystemError> = Self::__inner_create(
                     &info,
@@ -567,19 +601,11 @@ impl KernelThreadMechanism {
                 );
                 if result.is_err() {
                     // 创建失败
-                    info.created
-                        .store(KernelThreadCreateStatus::ErrorOccured, Ordering::SeqCst);
+                    info.set_create_error();
                 };
-                list = KTHREAD_CREATE_LIST.lock();
             }
-            drop(list);
 
             Self::reap_zombie_kthreads(&kthreadd_pcb);
-
-            let irq_guard = unsafe { CurrentIrqArch::save_and_disable_irq() };
-            ProcessManager::mark_sleep(true).ok();
-            drop(irq_guard);
-            schedule(SchedMode::SM_NONE);
         }
     }
 
@@ -606,6 +632,153 @@ impl KernelThreadMechanism {
     }
 }
 
+/// Exercise the externally visible kthread create/run/stop handshakes in a
+/// real DragonOS guest. This is intentionally invoked only through debugfs;
+/// production paths do not pay the stress-test cost.
+pub(crate) fn run_debug_selftests() -> Result<String, SystemError> {
+    if KTHREAD_SELFTEST_RUNNING
+        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+        .is_err()
+    {
+        return Err(SystemError::EBUSY);
+    }
+    let _guard = KthreadSelftestGuard;
+
+    let mut report = String::new();
+    let mut failures = 0usize;
+
+    append_kthread_selftest_case(
+        &mut report,
+        "create_stopped_stop",
+        selftest_create_stopped_stop(),
+        &mut failures,
+    );
+    append_kthread_selftest_case(
+        &mut report,
+        "create_and_run_stop",
+        selftest_create_and_run_stop(),
+        &mut failures,
+    );
+
+    let (quick_exit_ok, quick_exit_completed) = selftest_quick_exit(512);
+    append_kthread_selftest_case(&mut report, "quick_exit_512", quick_exit_ok, &mut failures);
+    report.push_str(&alloc::format!(
+        "quick_exit_completed={quick_exit_completed}\n"
+    ));
+
+    if failures == 0 {
+        report.insert_str(0, "status=ok\n");
+    } else {
+        report.insert_str(0, &alloc::format!("status=fail failures={failures}\n"));
+    }
+    Ok(report)
+}
+
+fn selftest_create_stopped_stop() -> bool {
+    let entered = Arc::new(AtomicUsize::new(0));
+    let entered_worker = entered.clone();
+    let name = "kthread-selftest-stopped".to_string();
+    let closure = KernelThreadClosure::EmptyClosure((
+        Box::new(move || {
+            entered_worker.fetch_add(1, Ordering::AcqRel);
+            0
+        }),
+        (),
+    ));
+    let Some(pcb) = KernelThreadMechanism::create(closure, name.clone()) else {
+        return false;
+    };
+
+    let worker_private_ready = pcb
+        .worker_private()
+        .as_ref()
+        .and_then(|private| private.kernel_thread())
+        .is_some();
+    let ready_ok = entered.load(Ordering::Acquire) == 0
+        && pcb.sched_info().state() == ProcessState::Blocked(false)
+        && pcb.flags().contains(ProcessFlags::KTHREAD)
+        && pcb.basic().name() == name
+        && worker_private_ready;
+    let stop_ok = KernelThreadMechanism::stop(&pcb).is_ok();
+
+    ready_ok && stop_ok && entered.load(Ordering::Acquire) == 0
+}
+
+fn selftest_create_and_run_stop() -> bool {
+    const RESULT: i32 = 73;
+
+    let entered = Arc::new(AtomicUsize::new(0));
+    let entered_worker = entered.clone();
+    let entered_completion = Arc::new(Completion::new());
+    let entered_completion_worker = entered_completion.clone();
+    let closure = KernelThreadClosure::EmptyClosure((
+        Box::new(move || {
+            entered_worker.fetch_add(1, Ordering::AcqRel);
+            entered_completion_worker.complete();
+            let current = ProcessManager::current_pcb();
+            while !KernelThreadMechanism::should_stop(&current) {
+                schedule(SchedMode::SM_NONE);
+            }
+            RESULT
+        }),
+        (),
+    ));
+    let Some(pcb) =
+        KernelThreadMechanism::create_and_run(closure, "kthread-selftest-running".to_string())
+    else {
+        return false;
+    };
+
+    if entered_completion.wait_for_completion().is_err() {
+        let _ = KernelThreadMechanism::stop(&pcb);
+        return false;
+    }
+    let result = KernelThreadMechanism::stop(&pcb);
+    entered.load(Ordering::Acquire) == 1 && result == Ok(RESULT as usize)
+}
+
+fn selftest_quick_exit(iterations: usize) -> (bool, usize) {
+    const RESULT: i32 = 91;
+
+    let completed = Arc::new(AtomicUsize::new(0));
+    for _ in 0..iterations {
+        let entered_completion = Arc::new(Completion::new());
+        let entered_completion_worker = entered_completion.clone();
+        let completed_worker = completed.clone();
+        let closure = KernelThreadClosure::EmptyClosure((
+            Box::new(move || {
+                completed_worker.fetch_add(1, Ordering::AcqRel);
+                entered_completion_worker.complete();
+                RESULT
+            }),
+            (),
+        ));
+        let Some(pcb) = KernelThreadMechanism::create_and_run(
+            closure,
+            "kthread-selftest-quick-exit".to_string(),
+        ) else {
+            return (false, completed.load(Ordering::Acquire));
+        };
+        if entered_completion.wait_for_completion().is_err()
+            || KernelThreadMechanism::stop(&pcb) != Ok(RESULT as usize)
+        {
+            return (false, completed.load(Ordering::Acquire));
+        }
+    }
+
+    let completed = completed.load(Ordering::Acquire);
+    (completed == iterations, completed)
+}
+
+fn append_kthread_selftest_case(report: &mut String, name: &str, ok: bool, failures: &mut usize) {
+    if ok {
+        report.push_str(&alloc::format!("{name}=ok\n"));
+    } else {
+        *failures += 1;
+        report.push_str(&alloc::format!("{name}=fail\n"));
+    }
+}
+
 /// 内核线程启动的第二阶段
 ///
 /// 该函数只能被`kernel_thread_bootstrap_stage1`调用（jmp到该函数）
@@ -615,19 +788,33 @@ impl KernelThreadMechanism {
 /// - ptr: 传入的参数，是一个指向`Arc<KernelThreadCreateInfo>`的指针
 pub unsafe extern "C" fn kernel_thread_bootstrap_stage2(ptr: *const KernelThreadCreateInfo) -> ! {
     let info = KernelThreadCreateInfo::parse_unsafe_arc_ptr(ptr);
+    let current = ProcessManager::current_pcb();
+
+    // Complete all thread-visible setup before publishing Created. The arch
+    // fork path may already have scheduled this child, so post-fork setup in
+    // the parent cannot provide this ordering guarantee.
+    current.set_name(info.name().clone());
+    info.setup_pcb(&current);
 
     let closure: Box<KernelThreadClosure> = info.take_closure().unwrap();
-    info.set_create_ok(ProcessManager::current_pcb());
     let to_mark_sleep = info.to_mark_sleep();
-    drop(info);
 
     if to_mark_sleep {
-        // 进入睡眠状态
+        // Match Linux kthread(): publish the stopped state before publishing
+        // the creation result. A create_and_run wake racing with schedule()
+        // then either keeps this current task runnable or re-enqueues it after
+        // schedule has dequeued it; it can no longer be lost.
         let irq_guard = CurrentIrqArch::save_and_disable_irq();
-        ProcessManager::mark_sleep(true).expect("Failed to mark sleep");
+        ProcessManager::mark_sleep(false).expect("Failed to mark sleep");
+        info.set_create_ok(current.clone());
+        drop(info);
         drop(irq_guard);
         schedule(SchedMode::SM_NONE);
+    } else {
+        info.set_create_ok(current.clone());
+        drop(info);
     }
+    drop(current);
 
     let mut retval = SystemError::EINTR.to_posix_errno();
 

kernel/src/process/manager/exit.rs

@@ -21,6 +21,7 @@ use crate::{
     },
     mm::IDLE_PROCESS_ADDRESS_SPACE,
     process::{
+        kthread::KernelThreadMechanism,
         pid::{Pid, PidType},
         ptrace, ProcessControlBlock, ProcessFlags, ProcessManager, ProcessState, RawPid,
     },
@@ -136,7 +137,7 @@ impl ProcessManager {
             // Explicitly wake kthreadd when a kthread exits so that it can
             // reap the zombie.
             if is_kthread {
-                let _ = ProcessManager::wakeup(&parent_pcb);
+                KernelThreadMechanism::notify_daemon();
             }
 
             // TODO: The signal delivery decision should also consider thread-group

user/apps/tests/dunitest/suites/fuse/fuse_test_simplefs_local.h

@@ -720,6 +720,7 @@ struct fuse_daemon_args {
     uint32_t root_open_out_flags;
     uint32_t hello_open_out_flags;
     volatile uint32_t *dynamic_hello_open_out_flags;
+    volatile uint32_t *dynamic_open_out_flags;
     volatile unsigned char *dynamic_hello_first_byte;
     volatile size_t *dynamic_hello_read_size;
     volatile size_t *dynamic_hello_byte_offset;
@@ -736,12 +737,14 @@ struct fuse_daemon_args {
     volatile uint32_t *access_count;
     volatile uint32_t *lookup_count;
     volatile uint32_t *flush_count;
+    volatile uint32_t *write_count_at_flush;
     volatile uint32_t *last_flush_uid;
     volatile uint32_t *last_flush_gid;
     volatile uint32_t *last_flush_pid;
     volatile uint32_t *fsync_count;
     volatile uint32_t *fsyncdir_count;
     volatile uint32_t *create_count;
+    volatile uint64_t *last_create_nodeid;
     volatile uint32_t *mknod_count;
     volatile uint32_t *rename2_count;
     volatile uint32_t *open_count;
@@ -757,6 +760,9 @@ struct fuse_daemon_args {
     volatile uint64_t *last_setattr_fh;
     volatile uint64_t *last_setattr_size;
     volatile uint64_t *last_setattr_lock_owner;
+    // UINT64_MAX means no override.  Tests use this to model a daemon-side
+    // size shrink observed by GETATTR without issuing a local SETATTR first.
+    volatile uint64_t *getattr_size_override;
     volatile uint64_t *last_fallocate_fh;
     volatile uint64_t *last_fallocate_offset;
     volatile uint64_t *last_fallocate_length;
@@ -790,7 +796,17 @@ struct fuse_daemon_args {
     volatile unsigned char *write_watch_bytes;
     volatile unsigned char *write_covers_watch;
     volatile unsigned char *backend_watch_byte;
+    unsigned char *large_write_backing;
+    size_t large_write_backing_capacity;
+    volatile uint64_t *large_write_nodeid;
     uint32_t write_trace_capacity;
+    volatile int *forced_write_errno;
+    volatile uint64_t *forced_write_offset;
+    volatile uint64_t *forced_short_write_offset;
+    volatile uint32_t *forced_short_write_size;
+    volatile uint64_t *block_write_offset;
+    volatile int *write_entered;
+    volatile int *release_write;
     volatile uint64_t *last_fsync_fh;
     volatile uint32_t *write_count_at_fsync;
     volatile uint32_t *last_write_flags_at_fsync;
@@ -1054,6 +1070,10 @@ static inline int fuse_handle_one(struct fuse_daemon_args *a, const unsigned cha
         struct fuse_attr_out out;
         memset(&out, 0, sizeof(out));
         simplefs_fill_attr(node, &out.attr);
+        if (a->getattr_size_override && *a->getattr_size_override != UINT64_MAX) {
+            out.attr.size = *a->getattr_size_override;
+            out.attr.blocks = (out.attr.size + 511) / 512;
+        }
         return fuse_write_reply(a->fd, h->unique, 0, &out, sizeof(out));
     }
     case FUSE_OPENDIR:
@@ -1093,7 +1113,10 @@ static inline int fuse_handle_one(struct fuse_daemon_args *a, const unsigned cha
         if (h->opcode == FUSE_OPEN && (in->flags & O_TRUNC)) {
             node->size = 0;
         }
-        if (h->opcode == FUSE_OPEN && h->nodeid == 2 && a->dynamic_hello_open_out_flags) {
+        if (h->opcode == FUSE_OPEN && a->dynamic_open_out_flags) {
+            node->open_out_flags = *a->dynamic_open_out_flags;
+        } else if (h->opcode == FUSE_OPEN && h->nodeid == 2 &&
+                   a->dynamic_hello_open_out_flags) {
             node->open_out_flags = *a->dynamic_hello_open_out_flags;
         }
         struct fuse_open_out out;
@@ -1167,6 +1190,14 @@ static inline int fuse_handle_one(struct fuse_daemon_args *a, const unsigned cha
             return fuse_write_reply(a->fd, h->unique, -forced_errno, NULL, 0);
         }
         size_t effective_size = node->size;
+        const unsigned char *read_backing = node->data;
+        if (a->large_write_backing && a->large_write_nodeid &&
+            *a->large_write_nodeid == h->nodeid) {
+            effective_size = node->size < a->large_write_backing_capacity
+                                 ? node->size
+                                 : a->large_write_backing_capacity;
+            read_backing = a->large_write_backing;
+        }
         int generated_hello = h->nodeid == 2 && a->hello_generated_size_override > 0;
         if (generated_hello) {
             effective_size = a->hello_generated_size_override;
@@ -1253,7 +1284,7 @@ static inline int fuse_handle_one(struct fuse_daemon_args *a, const unsigned cha
             free(generated);
             return ret;
         }
-        return fuse_write_reply(a->fd, h->unique, 0, node->data + in->offset, to_copy);
+        return fuse_write_reply(a->fd, h->unique, 0, read_backing + in->offset, to_copy);
     }
     case FUSE_READDIR:
     case FUSE_READDIRPLUS: {
@@ -1452,6 +1483,9 @@ static inline int fuse_handle_one(struct fuse_daemon_args *a, const unsigned cha
             return -1;
         }
         const struct fuse_flush_in *in = (const struct fuse_flush_in *)payload;
+        if (a->write_count_at_flush && a->write_count) {
+            *a->write_count_at_flush = *a->write_count;
+        }
         if (a->flush_count) {
             (*a->flush_count)++;
         }
@@ -1541,7 +1575,25 @@ static inline int fuse_handle_one(struct fuse_daemon_args *a, const unsigned cha
         if (!node || simplefs_node_is_dir(node) || simplefs_node_is_symlink(node)) {
             return fuse_write_reply(a->fd, h->unique, -EINVAL, NULL, 0);
         }
-        if (in->offset >= SIMPLEFS_DATA_MAX) {
+        unsigned char *write_backing = node->data;
+        size_t write_capacity = SIMPLEFS_DATA_MAX;
+        if (a->large_write_backing && a->large_write_nodeid &&
+            ((*a->large_write_nodeid == h->nodeid) ||
+             (*a->large_write_nodeid == 0 &&
+              in->offset + in->size > SIMPLEFS_DATA_MAX))) {
+            if (*a->large_write_nodeid == 0 && in->offset > 0) {
+                size_t prefix = (size_t)in->offset;
+                if (prefix > SIMPLEFS_DATA_MAX)
+                    prefix = SIMPLEFS_DATA_MAX;
+                if (prefix > a->large_write_backing_capacity)
+                    prefix = a->large_write_backing_capacity;
+                memcpy(a->large_write_backing, node->data, prefix);
+            }
+            *a->large_write_nodeid = h->nodeid;
+            write_backing = a->large_write_backing;
+            write_capacity = a->large_write_backing_capacity;
+        }
+        if (in->offset >= write_capacity) {
             return fuse_write_reply(a->fd, h->unique, -EFBIG, NULL, 0);
         }
         uint32_t write_index = 0;
@@ -1573,8 +1625,8 @@ static inline int fuse_handle_one(struct fuse_daemon_args *a, const unsigned cha
             *a->last_write_pid = h->pid;
         }
         size_t to_copy = in->size;
-        if (in->offset + to_copy > SIMPLEFS_DATA_MAX) {
-            to_copy = SIMPLEFS_DATA_MAX - (size_t)in->offset;
+        if (in->offset + to_copy > write_capacity) {
+            to_copy = write_capacity - (size_t)in->offset;
         }
         unsigned char watch_byte = 0;
         int covers_watch = 0;
@@ -1604,14 +1656,36 @@ static inline int fuse_handle_one(struct fuse_daemon_args *a, const unsigned cha
                 a->write_covers_watch[write_index] = covers_watch ? 1 : 0;
             }
         }
-        memcpy(node->data + in->offset, data, to_copy);
+        if (a->block_write_offset && a->write_entered && a->release_write &&
+            in->offset == *a->block_write_offset && !*a->release_write) {
+            __sync_synchronize();
+            *a->write_entered = 1;
+            __sync_synchronize();
+            while (!*a->release_write && (!a->stop || !*a->stop)) {
+                usleep(1000);
+            }
+        }
+        if (a->forced_write_errno && a->forced_write_offset &&
+            *a->forced_write_errno > 0 && in->offset == *a->forced_write_offset) {
+            if (a->write_count) {
+                __sync_synchronize();
+                (*a->write_count)++;
+            }
+            return fuse_write_reply(a->fd, h->unique, -*a->forced_write_errno, NULL, 0);
+        }
+        if (a->forced_short_write_offset && a->forced_short_write_size &&
+            in->offset == *a->forced_short_write_offset &&
+            *a->forced_short_write_size < to_copy) {
+            to_copy = *a->forced_short_write_size;
+        }
+        memcpy(write_backing + in->offset, data, to_copy);
         if (node->size < in->offset + to_copy) {
             node->size = (size_t)in->offset + to_copy;
         }
         if (a->backend_watch_byte) {
             uint64_t watch = a->write_watch_offset;
-            if (watch < node->size && watch < SIMPLEFS_DATA_MAX) {
-                *a->backend_watch_byte = node->data[watch];
+            if (watch < node->size && watch < write_capacity) {
+                *a->backend_watch_byte = write_backing[watch];
             }
         }
         if (a->write_count) {
@@ -1673,6 +1747,9 @@ static inline int fuse_handle_one(struct fuse_daemon_args *a, const unsigned cha
         strncpy(nnode->name, name, sizeof(nnode->name) - 1);
         nnode->name[sizeof(nnode->name) - 1] = '\0';
         nnode->size = 0;
+        if (a->last_create_nodeid) {
+            *a->last_create_nodeid = nnode->nodeid;
+        }
 
         struct {
             struct fuse_entry_out entry;

user/apps/tests/dunitest/suites/normal/fuse_stats_debugfs.cc

@@ -9,6 +9,7 @@
 #include <sys/stat.h>
 #include <unistd.h>
 
+#include <algorithm>
 #include <string>
 
 #include "../fuse/fuse_gtest_common.h"
@@ -163,14 +164,20 @@ TEST(FuseStatsDebugFs, StatsFileExistsAndSupportsOffsetReads) {
     expect_field(whole, "light direct_read_dma,read_size_buckets\n");
     expect_field(whole, "init_epoch ");
     expect_field(whole, "negotiated_max_read_bytes ");
+    expect_field(whole, "negotiated_max_write_bytes ");
     expect_field(whole, "negotiated_max_pages ");
     expect_field(whole, "negotiated_max_readahead_bytes ");
     expect_field(whole, "negotiated_async_read ");
+    expect_field(whole, "negotiated_writeback_cache ");
     expect_field(whole, "effective_read_payload_limit_bytes ");
+    expect_field(whole, "effective_write_payload_limit_bytes ");
     expect_field(whole, "request_queue_current ");
     expect_field(whole, "dispatch_current ");
     expect_field(whole, "processing_current ");
     expect_field(whole, "read_reservation_current ");
+    expect_field(whole, "invalidation_launder_batches_total ");
+    expect_field(whole, "invalidation_launder_pages_total ");
+    expect_field(whole, "mmap_writeback_admission_retries_total ");
     expect_field(whole, "requests_queued_total ");
     expect_field(whole, "requests_dequeued_total ");
     expect_field(whole, "requests_replied_ok_total ");
@@ -191,6 +198,18 @@ TEST(FuseStatsDebugFs, StatsFileExistsAndSupportsOffsetReads) {
     EXPECT_EQ(0, parse_counter(after_fuse, "dispatch_current"));
     EXPECT_EQ(0, parse_counter(after_fuse, "processing_current"));
     EXPECT_EQ(0, parse_counter(after_fuse, "read_reservation_current"));
+    const long long negotiated_max_write =
+        parse_counter(after_fuse, "negotiated_max_write_bytes");
+    const long long negotiated_max_pages =
+        parse_counter(after_fuse, "negotiated_max_pages");
+    ASSERT_GE(negotiated_max_write, 4096);
+    ASSERT_GE(negotiated_max_pages, 1);
+    EXPECT_LE(negotiated_max_write, 1024 * 1024);
+    EXPECT_EQ(0, parse_counter(after_fuse, "negotiated_writeback_cache"));
+    const long long expected_write_pages =
+        std::min(64LL, std::min(negotiated_max_pages, negotiated_max_write / 4096));
+    EXPECT_EQ(expected_write_pages * 4096,
+              parse_counter(after_fuse, "effective_write_payload_limit_bytes"));
 
     expect_field(whole, "[virtiofs]\n");
     expect_field(whole, "device_queue_depth_max ");

user/apps/tests/dunitest/suites/normal/kthread_selftest.cc

@@ -0,0 +1,58 @@
+#include <gtest/gtest.h>
+
+#include <errno.h>
+#include <fcntl.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <string>
+
+namespace {
+
+constexpr const char* kKthreadSelftestPath = "/sys/kernel/debug/kthread/selftest";
+
+std::string ReadAll(const char* path) {
+    int fd = open(path, O_RDONLY);
+    EXPECT_GE(fd, 0) << "open(" << path << ") failed: errno=" << errno << " (" << strerror(errno)
+                     << ")";
+    if (fd < 0) {
+        return {};
+    }
+
+    std::string content;
+    char buf[256];
+    while (true) {
+        ssize_t n = read(fd, buf, sizeof(buf));
+        if (n == 0) {
+            break;
+        }
+        EXPECT_GT(n, 0) << "read(" << path << ") failed: errno=" << errno << " ("
+                        << strerror(errno) << ")";
+        if (n <= 0) {
+            close(fd);
+            return {};
+        }
+        content.append(buf, static_cast<size_t>(n));
+    }
+
+    EXPECT_EQ(0, close(fd)) << "close(" << path << ") failed: errno=" << errno << " ("
+                            << strerror(errno) << ")";
+    return content;
+}
+
+}  // namespace
+
+TEST(KthreadSelftest, CreateRunStopAndReapHandshakes) {
+    const std::string report = ReadAll(kKthreadSelftestPath);
+    ASSERT_FALSE(report.empty());
+    EXPECT_NE(std::string::npos, report.find("status=ok\n")) << report;
+    EXPECT_NE(std::string::npos, report.find("create_stopped_stop=ok\n")) << report;
+    EXPECT_NE(std::string::npos, report.find("create_and_run_stop=ok\n")) << report;
+    EXPECT_NE(std::string::npos, report.find("quick_exit_512=ok\n")) << report;
+    EXPECT_NE(std::string::npos, report.find("quick_exit_completed=512\n")) << report;
+}
+
+int main(int argc, char** argv) {
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}

user/apps/tests/dunitest/whitelist.txt

@@ -32,6 +32,7 @@ normal/sched_affinity
 normal/sync_file_range
 normal/splice_concurrent_io
 normal/rcu_selftest
+normal/kthread_selftest
 normal/restart_syscall_semantics
 normal/errseq_selftest
 normal/errseq_writeback_reporting

user/apps/virtiofs_bench/transcript_test.sh

@@ -46,7 +46,24 @@ check_transcript() {
                 (value("status") != "ok" && value("status") != "fail") ||
                 !decimal("errno") || !decimal("elapsed_us") || !decimal("bytes") ||
                 !decimal("ops") || !decimal("syscalls") || !decimal("short_io") ||
-                !decimal("eintr") || checksum !~ /^[0-9a-f]+$/ || length(checksum) != 16) {
+                !decimal("eintr") || !decimal("data_loop_us") || !decimal("fsync_us") ||
+                !decimal("close_us") || !decimal("end_to_end_us") ||
+                checksum !~ /^[0-9a-f]+$/ || length(checksum) != 16) {
+                bad = 1
+            }
+            workload = value("workload")
+            data_loop = value("data_loop_us") + 0
+            fsync_time = value("fsync_us") + 0
+            close_time = value("close_us") + 0
+            end_to_end = value("end_to_end_us") + 0
+            elapsed = value("elapsed_us") + 0
+            if (workload == "sequential_write" || workload == "prepare") {
+                if (data_loop == 0 || end_to_end == 0 || elapsed != data_loop ||
+                    end_to_end < data_loop + fsync_time + close_time) {
+                    bad = 1
+                }
+            } else if (data_loop != 0 || fsync_time != 0 || close_time != 0 ||
+                       end_to_end != 0) {
                 bad = 1
             }
         }
@@ -54,6 +71,21 @@ check_transcript() {
     '
 }
 
+check_result_only_transcript() {
+    awk '
+        $1 == "phase" { phases++ }
+        $1 == "result" {
+            results++
+            for (i = 1; i <= NF; ++i) {
+                if ($i ~ /^(data_loop_us|fsync_us|close_us|end_to_end_us)=[0-9]+$/) {
+                    timings++
+                }
+            }
+        }
+        END { exit phases != 0 || results != 1 || timings != 4 }
+    '
+}
+
 MOUNT="$WORK_DIR/mount"
 mkdir "$MOUNT"
 
@@ -62,11 +94,35 @@ VIRTIOFS_BENCH_RUN_ID=host_prepare VIRTIOFS_BENCH_CACHE_MODE=warm \
     --file-size 16384 --block-size 4096 >"$WORK_DIR/prepare.log" 2>&1
 check_transcript <"$WORK_DIR/prepare.log"
 
+VIRTIOFS_BENCH_PHASE_MARKERS=0 VIRTIOFS_BENCH_RUN_ID=host_perf \
+    "$BIN" --mount "$MOUNT" --workload prepare --path perf_schema_test \
+    --file-size 16384 --block-size 4096 >"$WORK_DIR/perf.log" 2>&1
+check_result_only_transcript <"$WORK_DIR/perf.log"
+
 VIRTIOFS_BENCH_RUN_ID=host_read VIRTIOFS_BENCH_CACHE_MODE=warm \
     "$BIN" --mount "$MOUNT" --workload sequential_read --path schema_test \
     --file-size 16384 --block-size 4096 >"$WORK_DIR/read.log" 2>&1
 check_transcript <"$WORK_DIR/read.log"
 
+VIRTIOFS_BENCH_RUN_ID=host_write VIRTIOFS_BENCH_CACHE_MODE=warm \
+    "$BIN" --mount "$MOUNT" --workload sequential_write --path schema_test \
+    --file-size 16384 --block-size 4096 >"$WORK_DIR/write.log" 2>&1
+check_transcript <"$WORK_DIR/write.log"
+
+# Timing fields are a semantic schema, not just four decimal strings.
+sed 's/ data_loop_us=[^ ]*/ data_loop_us=1/' "$WORK_DIR/read.log" \
+    >"$WORK_DIR/malformed-read-timing.log"
+if check_transcript <"$WORK_DIR/malformed-read-timing.log"; then
+    echo "non-write timing unexpectedly accepted" >&2
+    exit 1
+fi
+sed 's/ elapsed_us=[^ ]*/ elapsed_us=0/' "$WORK_DIR/write.log" \
+    >"$WORK_DIR/malformed-write-timing.log"
+if check_transcript <"$WORK_DIR/malformed-write-timing.log"; then
+    echo "inconsistent write timing unexpectedly accepted" >&2
+    exit 1
+fi
+
 # A missing mandatory result field must be rejected by the host parser.
 sed 's/ syscalls=[^ ]*//' "$WORK_DIR/read.log" >"$WORK_DIR/malformed.log"
 if check_transcript <"$WORK_DIR/malformed.log"; then

user/apps/virtiofs_bench/virtiofs_bench.cc

@@ -539,12 +539,21 @@ struct IoCounters {
     uint64_t eintr = 0;
 };
 
+struct WritePhaseTimings {
+    uint64_t data_loop_us = 0;
+    uint64_t fsync_us = 0;
+    uint64_t close_us = 0;
+    uint64_t end_to_end_us = 0;
+};
+
 void emit_result(const char* workload, const Options& opt, uint64_t elapsed_us, uint64_t bytes,
-                 uint64_t ops, int err, const IoCounters& io = {}, uint64_t checksum = 0) {
+                 uint64_t ops, int err, const IoCounters& io = {}, uint64_t checksum = 0,
+                 const WritePhaseTimings& write_timings = {}) {
     utsname uts = {};
     uname(&uts);
     printf("result workload=%s status=%s errno=%d elapsed_us=%llu bytes=%llu ops=%llu "
            "syscalls=%llu short_io=%llu eintr=%llu checksum=%016llx "
+           "data_loop_us=%llu fsync_us=%llu close_us=%llu end_to_end_us=%llu "
            "mount=%s dataset=%s seed=%llu files=%zu file_size=%zu block_size=%zu "
            "iterations=%zu workers=%zu run_id=%s "
            "cache_mode=%s mount_options=%s expect_dax=%s sysname=%s release=%s\n",
@@ -555,7 +564,11 @@ void emit_result(const char* workload, const Options& opt, uint64_t elapsed_us,
            static_cast<unsigned long long>(io.syscalls),
            static_cast<unsigned long long>(io.short_io),
            static_cast<unsigned long long>(io.eintr),
-           static_cast<unsigned long long>(checksum), opt.mount.c_str(),
+           static_cast<unsigned long long>(checksum),
+           static_cast<unsigned long long>(write_timings.data_loop_us),
+           static_cast<unsigned long long>(write_timings.fsync_us),
+           static_cast<unsigned long long>(write_timings.close_us),
+           static_cast<unsigned long long>(write_timings.end_to_end_us), opt.mount.c_str(),
            opt.path.empty() ? "ephemeral" : opt.path.c_str(),
            static_cast<unsigned long long>(opt.seed), opt.files, opt.file_size, opt.block_size,
            opt.iterations, opt.workers, env_or_empty("VIRTIOFS_BENCH_RUN_ID"),
@@ -640,6 +653,10 @@ struct DatasetManifest {
 
 void phase_marker(const Options& opt, const char* workload, const char* phase, const char* event,
                   uint64_t offset, uint64_t requested, int64_t returned, int err) {
+    const char* enabled = getenv("VIRTIOFS_BENCH_PHASE_MARKERS");
+    if (enabled && strcmp(enabled, "0") == 0) {
+        return;
+    }
     // DragonOS's current stderr formatter truncates at 64-bit printf
     // conversions. Build the record from decimal strings so each phase remains
     // a single, shell-tokenizable line without losing timing or I/O context.
@@ -1089,9 +1106,11 @@ int sequential_write_phase(const Options& opt, WorkloadSpec workload) {
     uint64_t bytes = 0;
     IoCounters io;
     uint64_t elapsed_us = 0;
+    WritePhaseTimings write_timings;
     if (fd >= 0) {
         phase_marker(opt, label, "data_loop", "begin", 0, opt.file_size, 0, 0);
-        uint64_t start = now_us();
+        const uint64_t end_to_end_start = now_us();
+        uint64_t start = end_to_end_start;
         while (err == 0 && bytes < opt.file_size) {
             size_t requested = std::min(opt.block_size, opt.file_size - static_cast<size_t>(bytes));
             if (!write_all_counted(fd, data.data() + bytes, requested, &io)) {
@@ -1101,16 +1120,23 @@ int sequential_write_phase(const Options& opt, WorkloadSpec workload) {
             bytes += requested;
         }
         elapsed_us = now_us() - start;
+        write_timings.data_loop_us = elapsed_us;
         phase_marker(opt, label, "data_loop", "end", bytes, 0,
                      err == 0 ? static_cast<int64_t>(bytes) : -1, err);
 
         if (err == 0) {
             phase_marker(opt, label, "fsync", "begin", bytes, 0, 0, 0);
+            start = now_us();
             fsync_preserve_error(fd, &err);
+            write_timings.fsync_us = now_us() - start;
             phase_marker(opt, label, "fsync", "end", bytes, 0, err == 0 ? 0 : -1, err);
         }
         phase_marker(opt, label, "close", "begin", bytes, 0, 0, 0);
+        start = now_us();
         close_preserve_error(fd, &err);
+        const uint64_t close_end = now_us();
+        write_timings.close_us = close_end - start;
+        write_timings.end_to_end_us = close_end - end_to_end_start;
         phase_marker(opt, label, "close", "end", bytes, 0, err == 0 ? 0 : -1, err);
     }
 
@@ -1127,7 +1153,7 @@ int sequential_write_phase(const Options& opt, WorkloadSpec workload) {
         unlinkat(root_fd, manifest_temp.c_str(), 0);
     }
     close(root_fd);
-    emit_result(label, opt, elapsed_us, bytes, io.syscalls, err, io, checksum);
+    emit_result(label, opt, elapsed_us, bytes, io.syscalls, err, io, checksum, write_timings);
     emit_io_summary(label, io, checksum, 0);
     return err == 0 ? 0 : -1;
 }