feat(fs): partial work for ext4's page cache

Fix page cache's bug, add size check in read function.

Add page cache's base operations for ext4, but the cachepage will not be
dropped until kernel stop, so we need to call fsync function manually,
consider use some strategy such as LRU.

src/fs/ext4.rs

@@ -1,6 +1,6 @@
 use core::sync::atomic::{AtomicU32, AtomicU64, Ordering};
 
-use crate::kernel::mem::{PageCache, PageCacheBackend};
+use crate::kernel::mem::{CachePage, CachePageStream, PageCache, PageCacheBackend};
 use crate::{
     io::{Buffer, ByteBuffer, Stream},
     kernel::{
@@ -83,7 +83,7 @@ impl Vfs for Ext4Fs {
     }
 
     fn is_read_only(&self) -> bool {
-        true
+        false
     }
 }
 
@@ -257,12 +257,12 @@ impl FileInode {
 }
 
 impl PageCacheBackend for FileInode {
-    fn read_page(&self, page: &mut crate::kernel::mem::CachePage, offset: usize) -> KResult<usize> {
+    fn read_page(&self, page: &mut CachePage, offset: usize) -> KResult<usize> {
         self.read_direct(page, offset)
     }
 
-    fn write_page(&self, page: &crate::kernel::mem::CachePage, offset: usize) -> KResult<usize> {
-        todo!()
+    fn write_page(&self, page: &mut CachePageStream, offset: usize) -> KResult<usize> {
+        self.write_direct(page, offset)
     }
 
     fn size(&self) -> usize {
@@ -296,12 +296,6 @@ impl Inode for FileInode {
     fn write(&self, stream: &mut dyn Stream, offset: WriteOffset) -> KResult<usize> {
         let _lock = Task::block_on(self.rwsem.write());
 
-        let vfs = self.vfs.upgrade().ok_or(EIO)?;
-        let ext4fs = vfs.as_any().downcast_ref::<Ext4Fs>().unwrap();
-
-        let mut temp_buf = vec![0u8; 4096];
-        let mut total_written = 0;
-
         let mut store_new_end = None;
         let offset = match offset {
             WriteOffset::Position(offset) => offset,
@@ -312,6 +306,31 @@ impl Inode for FileInode {
             }
         };
 
+        let total_written = Task::block_on(self.page_cache.write(stream, offset))?;
+        let cursor_end = offset + total_written;
+        if let Some(store_end) = store_new_end {
+            *store_end = cursor_end;
+        }
+
+        let mtime = Instant::now();
+        *self.mtime.lock() = mtime;
+        self.size.store(cursor_end as u64, Ordering::Relaxed);
+
+        // TODO: change this with some update strategy such as LRU
+        let _ = Task::block_on(self.page_cache.fsync());
+
+        Ok(total_written)
+    }
+
+    fn write_direct(&self, stream: &mut dyn Stream, offset: usize) -> KResult<usize> {
+        //let _lock = Task::block_on(self.rwsem.write());
+
+        let vfs = self.vfs.upgrade().ok_or(EIO)?;
+        let ext4fs = vfs.as_any().downcast_ref::<Ext4Fs>().unwrap();
+
+        let mut temp_buf = vec![0u8; 4096];
+        let mut total_written = 0;
+
         while let Some(data) = stream.poll_data(&mut temp_buf)? {
             let written = ext4fs
                 .inner
@@ -323,18 +342,10 @@ impl Inode for FileInode {
             }
         }
 
-        if let Some(store_end) = store_new_end {
-            *store_end = offset + total_written;
-        }
-        let mtime = Instant::now();
-        *self.mtime.lock() = mtime;
-        let new_size = (offset + total_written) as u64;
-        self.size
-            .store(offset as u64 + total_written as u64, Ordering::Relaxed);
         ext4fs.modify_inode_stat(
             self.ino as u32,
-            Some(new_size),
-            mtime.since_epoch().as_secs() as u32,
+            Some(self.size() as u64),
+            self.mtime.lock().since_epoch().as_secs() as u32,
         );
 
         Ok(total_written)

src/fs/fat32.rs

@@ -3,7 +3,7 @@ mod file;
 
 use crate::io::Stream;
 use crate::kernel::constants::EIO;
-use crate::kernel::mem::AsMemoryBlock;
+use crate::kernel::mem::{AsMemoryBlock, CachePageStream};
 use crate::kernel::vfs::inode::WriteOffset;
 use crate::{
     io::{Buffer, ByteBuffer, UninitBuffer},
@@ -308,11 +308,11 @@ impl Inode for FileInode {
         Ok(buffer.wrote())
     }
 
-    fn write(&self, stream: &mut dyn Stream, offset: WriteOffset) -> KResult<usize> {
+    fn write(&self, _stream: &mut dyn Stream, _offset: WriteOffset) -> KResult<usize> {
         todo!()
     }
 
-    fn write_direct(&self, stream: &mut dyn Stream, offset: WriteOffset) -> KResult<usize> {
+    fn write_direct(&self, _stream: &mut dyn Stream, _offset: usize) -> KResult<usize> {
         todo!()
     }
 }
@@ -322,7 +322,7 @@ impl PageCacheBackend for FileInode {
         self.read_direct(page, offset)
     }
 
-    fn write_page(&self, page: &CachePage, offset: usize) -> KResult<usize> {
+    fn write_page(&self, _page: &mut CachePageStream, _offset: usize) -> KResult<usize> {
         todo!()
     }
 

src/fs/tmpfs.rs

@@ -1,6 +1,6 @@
 use crate::io::Stream;
 use crate::kernel::constants::{EEXIST, EINVAL, EIO, EISDIR, ENOENT, ENOSYS, ENOTDIR};
-use crate::kernel::mem::{CachePage, PageCache, PageCacheBackend};
+use crate::kernel::mem::{CachePage, CachePageStream, PageCache, PageCacheBackend};
 use crate::kernel::timer::Instant;
 use crate::kernel::vfs::inode::InodeData;
 use crate::kernel::vfs::inode::RenameData;
@@ -496,7 +496,7 @@ impl PageCacheBackend for FileInode {
         Ok(PAGE_SIZE)
     }
 
-    fn write_page(&self, _page: &CachePage, _offset: usize) -> KResult<usize> {
+    fn write_page(&self, _page: &mut CachePageStream, _offset: usize) -> KResult<usize> {
         Ok(PAGE_SIZE)
     }
 
@@ -511,13 +511,13 @@ impl Inode for FileInode {
     }
 
     fn read(&self, buffer: &mut dyn Buffer, offset: usize) -> KResult<usize> {
-        let lock = Task::block_on(self.rwsem.write());
+        let _lock = Task::block_on(self.rwsem.write());
         Task::block_on(self.pages.read(buffer, offset))
     }
 
     fn write(&self, stream: &mut dyn Stream, offset: WriteOffset) -> KResult<usize> {
         // TODO: We don't need that strong guarantee, find some way to avoid locks
-        let lock = Task::block_on(self.rwsem.write());
+        let _lock = Task::block_on(self.rwsem.write());
 
         let mut store_new_end = None;
         let offset = match offset {
@@ -545,7 +545,7 @@ impl Inode for FileInode {
     }
 
     fn truncate(&self, length: usize) -> KResult<()> {
-        let lock = Task::block_on(self.rwsem.write());
+        let _lock = Task::block_on(self.rwsem.write());
         Task::block_on(self.pages.resize(length))?;
         self.size.store(length as u64, Ordering::Relaxed);
         *self.mtime.lock() = Instant::now();

src/kernel/mem.rs

@@ -12,5 +12,5 @@ pub use access::{AsMemoryBlock, MemoryBlock, PhysAccess};
 pub(self) use mm_area::MMArea;
 pub use mm_list::{handle_kernel_page_fault, FileMapping, MMList, Mapping, Permission};
 pub use page_alloc::{GlobalPageAlloc, RawPage};
-pub use page_cache::{CachePage, PageCache, PageCacheBackend};
+pub use page_cache::{CachePage, CachePageStream, PageCache, PageCacheBackend};
 pub use paging::{Page, PageBuffer};

src/kernel/mem/page_cache.rs

@@ -125,27 +125,32 @@ impl PageCache {
 
     pub async fn read(&self, buffer: &mut dyn Buffer, mut offset: usize) -> KResult<usize> {
         let mut pages = self.pages.lock().await;
+        let size = self.backend.upgrade().unwrap().size();
 
         loop {
+            if offset >= size {
+                break;
+            }
             let page_id = offset >> PAGE_SIZE_BITS;
             let page = pages.get(&page_id);
 
             match page {
                 Some(page) => {
                     let inner_offset = offset % PAGE_SIZE;
+                    let available_in_file = size.saturating_sub(offset);
 
                     // TODO: still cause unnecessary IO if valid_size < PAGESIZE
                     //       and fill result is Done
-                    if page.valid_size() == 0
-                        || buffer
-                            .fill(&page.valid_data()[inner_offset..])?
-                            .should_stop()
+                    let page_data = &page.valid_data()[inner_offset..];
+                    let read_size = page_data.len().min(available_in_file);
+
+                    if read_size == 0
+                        || buffer.fill(&page_data[..read_size])?.should_stop()
                         || buffer.available() == 0
                     {
                         break;
                     }
-
-                    offset += PAGE_SIZE - inner_offset;
+                    offset += read_size;
                 }
                 None => {
                     let mut new_page = CachePage::new();
@@ -217,7 +222,7 @@ impl PageCache {
                 self.backend
                     .upgrade()
                     .unwrap()
-                    .write_page(page, page_id << PAGE_SIZE_BITS)?;
+                    .write_page(&mut CachePageStream::new(*page), page_id << PAGE_SIZE_BITS)?;
                 page.clear_dirty();
             }
         }
@@ -293,6 +298,51 @@ impl PageCache {
     }
 }
 
+pub struct CachePageStream {
+    page: CachePage,
+    cur: usize,
+}
+
+impl CachePageStream {
+    pub fn new(page: CachePage) -> Self {
+        Self { page, cur: 0 }
+    }
+
+    pub fn remaining(&self) -> usize {
+        self.page.valid_size().saturating_sub(self.cur)
+    }
+
+    pub fn is_drained(&self) -> bool {
+        self.cur >= self.page.valid_size()
+    }
+}
+
+impl Stream for CachePageStream {
+    fn poll_data<'a>(&mut self, buf: &'a mut [u8]) -> KResult<Option<&'a mut [u8]>> {
+        if self.cur >= self.page.valid_size() {
+            return Ok(None);
+        }
+
+        let page_data = &self.page.all()[self.cur..self.page.valid_size()];
+        let to_read = buf.len().min(page_data.len());
+
+        buf[..to_read].copy_from_slice(&page_data[..to_read]);
+        self.cur += to_read;
+
+        Ok(Some(&mut buf[..to_read]))
+    }
+
+    fn ignore(&mut self, len: usize) -> KResult<Option<usize>> {
+        if self.cur >= self.page.valid_size() {
+            return Ok(None);
+        }
+
+        let to_ignore = len.min(self.page.valid_size() - self.cur);
+        self.cur += to_ignore;
+        Ok(Some(to_ignore))
+    }
+}
+
 // with this trait, "page cache" and "block cache" are unified,
 // for fs, offset is file offset (floor algin to PAGE_SIZE)
 // for blkdev, offset is block idx (floor align to PAGE_SIZE / BLK_SIZE)
@@ -300,7 +350,7 @@ impl PageCache {
 pub trait PageCacheBackend {
     fn read_page(&self, page: &mut CachePage, offset: usize) -> KResult<usize>;
 
-    fn write_page(&self, page: &CachePage, offset: usize) -> KResult<usize>;
+    fn write_page(&self, page: &mut CachePageStream, offset: usize) -> KResult<usize>;
 
     fn size(&self) -> usize;
 }

src/kernel/vfs/inode.rs

@@ -136,7 +136,7 @@ pub trait Inode: Send + Sync + InodeInner + Any {
         Err(if self.is_dir() { EISDIR } else { EINVAL })
     }
 
-    fn write_direct(&self, stream: &mut dyn Stream, offset: WriteOffset) -> KResult<usize> {
+    fn write_direct(&self, stream: &mut dyn Stream, offset: usize) -> KResult<usize> {
         Err(if self.is_dir() { EISDIR } else { EINVAL })
     }