backport #1311 # 1348

rafs/src/fs.rs

@@ -54,6 +54,10 @@ pub type Handle = u64;
 pub const RAFS_DEFAULT_ATTR_TIMEOUT: u64 = 1 << 32;
 /// Rafs default entry timeout value.
 pub const RAFS_DEFAULT_ENTRY_TIMEOUT: u64 = RAFS_DEFAULT_ATTR_TIMEOUT;
+/// FUSE_KERN_BUF_SIZE * pagesize() + FUSE_HEADER_ SIZE
+/// Use a const as stable v2.1 does not complie with latset fuse-backend-rs that
+/// exports a writer's available buffer size.
+pub const RAFS_MAX_AMPLIFIED_IO: usize = 1052656;
 
 fn default_threads_count() -> usize {
     8
@@ -765,18 +769,19 @@ impl FileSystem for Rafs {
         debug_assert!(!descs.is_empty() && !descs[0].bi_vec.is_empty());
 
         // Try to amplify user io for Rafs v5, to improve performance.
-        if self.sb.meta.is_v5() && size < self.amplify_io {
+        let amplify_io = cmp::min(self.amplify_io as usize, RAFS_MAX_AMPLIFIED_IO) as u32;
+        if self.sb.meta.is_v5() && size < amplify_io {
             let all_chunks_ready = self.device.all_chunks_ready(&descs);
             if !all_chunks_ready {
-                let chunk_size = self.metadata().chunk_size as u64;
-                let next_chunk_base = (offset + (size as u64) + chunk_size) & !chunk_size;
-                let window_base = std::cmp::min(next_chunk_base, inode_size);
-                let actual_size = window_base - (offset & !chunk_size);
-                if actual_size < self.amplify_io as u64 {
-                    let window_size = self.amplify_io as u64 - actual_size;
+                let chunk_mask = self.metadata().chunk_size as u64 - 1;
+                let next_chunk_base = (offset + (size as u64) + chunk_mask) & !chunk_mask;
+                let window_base = cmp::min(next_chunk_base, inode_size);
+                let actual_size = window_base - (offset & !chunk_mask);
+                if actual_size < amplify_io as u64 {
+                    let window_size = amplify_io as u64 - actual_size;
                     self.sb.amplify_io(
                         &self.device,
-                        self.amplify_io,
+                        amplify_io,
                         &mut descs,
                         &inode,
                         window_base,
@@ -797,7 +802,7 @@ impl FileSystem for Rafs {
             let r = self.device.read_to(w, desc)?;
             result += r;
             recorder.mark_success(r);
-            if r as u32 != desc.bi_size {
+            if r as u64 != desc.bi_size {
                 break;
             }
         }

rafs/src/metadata/direct_v6.rs

@@ -1287,7 +1287,7 @@ impl RafsInode for OndiskInodeWrapper {
 
         let mut descs = BlobIoVec::new();
         descs.bi_vec.push(desc);
-        descs.bi_size += content_len;
+        descs.bi_size += content_len as u64;
         left -= content_len;
 
         if left != 0 {
@@ -1311,7 +1311,7 @@ impl RafsInode for OndiskInodeWrapper {
                 }
 
                 descs.bi_vec.push(desc);
-                descs.bi_size += content_len;
+                descs.bi_size += content_len as u64;
                 left -= content_len;
                 if left == 0 {
                     break;

rafs/src/metadata/layout/v5.rs

@@ -1324,7 +1324,7 @@ fn add_chunk_to_bio_desc(
         (chunk_end - chunk_start) as u32,
         user_io,
     );
-    desc.bi_size += bio.size;
+    desc.bi_size += bio.size as u64;
     desc.bi_vec.push(bio);
 
     true

storage/src/cache/worker.rs

@@ -186,9 +186,9 @@ impl AsyncWorkerMgr {
     }
 
     /// Consume network bandwidth budget for prefetching.
-    pub fn consume_prefetch_budget(&self, size: u32) {
+    pub fn consume_prefetch_budget(&self, size: u64) {
         if self.prefetch_inflight.load(Ordering::Relaxed) > 0 {
-            if let Some(v) = NonZeroU32::new(size) {
+            if let Some(v) = NonZeroU32::new(size as u32) {
                 // Try to consume budget but ignore result.
                 if let Some(limiter) = self.prefetch_limiter.as_ref() {
                     let _ = limiter.check_n(v);

storage/src/device.rs

@@ -543,7 +543,7 @@ pub struct BlobIoVec {
     /// Blob IO flags.
     pub bi_flags: u32,
     /// Total size of blob IOs to be performed.
-    pub bi_size: u32,
+    pub bi_size: u64,
     /// Array of blob IOs, these IOs should executed sequentially.
     // TODO: As bi_vec must stay within the same blob, move BlobInfo out here?
     pub bi_vec: Vec<BlobIoDesc>,
@@ -560,7 +560,7 @@ impl BlobIoVec {
     /// Append another blob io vector to current one.
     pub fn append(&mut self, mut desc: BlobIoVec) {
         self.bi_vec.append(desc.bi_vec.as_mut());
-        self.bi_size += desc.bi_size;
+        self.bi_size += desc.bi_size as u64;
         debug_assert!(self.validate());
     }