block: Drop the middle buffer on the unaligned O_DIRECT path

run_unaligned_operation staged every unaligned request in a plain Vec
and then handed it to AlignedFile, which bounced again through an
aligned buffer. That Vec only gave the operation a contiguous range to
scatter into or gather from, which the aligned buffer already is, so
each slow path request paid for an extra allocation and a full length
copy.

Add read_unaligned and write_unaligned on AlignedFile that own the
single aligned bounce and scatter or gather through a closure over the
staging slice. run_unaligned_operation and the FileExt read_at and
write_at impls both route through them, so the staging and
read-modify-write logic lives in one place. The closures keep
AlignedFile free of any AsyncIoOperation dependency.

Signed-off-by: Anatol Belski <anbelski@linux.microsoft.com>

block/src/aligned_file.rs

@@ -111,6 +111,33 @@ impl AlignedFile {
             position: 0,
         }
     }
+
+    /// Read `len` bytes at `offset` through an aligned bounce buffer.
+    pub(crate) fn read_unaligned(
+        &self,
+        offset: u64,
+        len: usize,
+        scatter: impl FnOnce(&[u8]) -> io::Result<()>,
+    ) -> io::Result<usize> {
+        let mut abuf = AlignedBuffer::new(offset, len, self.alignment)?;
+        let n = abuf.read_from(&self.file)?;
+        scatter(&abuf.as_slice()[..n])?;
+        Ok(n)
+    }
+
+    /// Write `len` bytes at `offset` through an aligned bounce buffer.
+    pub(crate) fn write_unaligned(
+        &self,
+        offset: u64,
+        len: usize,
+        gather: impl FnOnce(&mut [u8]) -> io::Result<()>,
+    ) -> io::Result<usize> {
+        let mut abuf = AlignedBuffer::new(offset, len, self.alignment)?;
+        abuf.read_from(&self.file)?; // RMW: preserve head/tail padding
+        gather(abuf.as_mut_slice())?;
+        abuf.write_to(&self.file)?;
+        Ok(len)
+    }
 }
 
 impl FileExt for AlignedFile {
@@ -121,10 +148,10 @@ impl FileExt for AlignedFile {
         if is_aligned(self.alignment, buf.as_ptr() as usize, buf.len(), offset) {
             return self.file.read_at(buf, offset);
         }
-        let mut abuf = AlignedBuffer::new(offset, buf.len(), self.alignment)?;
-        let n = abuf.read_from(&self.file)?;
-        buf[..n].copy_from_slice(&abuf.as_slice()[..n]);
-        Ok(n)
+        self.read_unaligned(offset, buf.len(), |data| {
+            buf[..data.len()].copy_from_slice(data);
+            Ok(())
+        })
     }
 
     fn write_at(&self, buf: &[u8], offset: u64) -> io::Result<usize> {
@@ -134,11 +161,10 @@ impl FileExt for AlignedFile {
         if is_aligned(self.alignment, buf.as_ptr() as usize, buf.len(), offset) {
             return self.file.write_at(buf, offset);
         }
-        let mut abuf = AlignedBuffer::new(offset, buf.len(), self.alignment)?;
-        abuf.read_from(&self.file)?; // RMW: preserve head/tail padding
-        abuf.as_mut_slice().copy_from_slice(buf);
-        abuf.write_to(&self.file)?;
-        Ok(buf.len())
+        self.write_unaligned(offset, buf.len(), |dst| {
+            dst.copy_from_slice(buf);
+            Ok(())
+        })
     }
 }