Fix stat to use the fd instead of the path (#649)

* chore(deps): Upgrade to flume 0.11

Has a race condition fix for async APIs. Not sure if this actually
impacts us but sounds like a worthwhile upgrade.

* Add failing test cases (#648)

Attempting to stat on an unlinked file will error with NotFound when it
shouldn't since we still have the fd. Similarly, invoking stat on a
tempfile will stat the parent directory instead of the file itself.

* Fix stat() behavior to use the fd instead of the path

This fixes the broken test cases - unlinked files will still stat
successfully and temporary files will return stat for the file rather
than the parent directory.

There's probably a negligible performance boost since we don't need to
do anything with the path (I believe it saves a memory allocation + some
other cycles) + the kernel might have to do less work handling the
syscall. Of course, stat should never be in the hot path so it's largely
an academic difference.

glommio/Cargo.toml

@@ -26,7 +26,7 @@ buddy-alloc = "0.4"
 concurrent-queue = "1.2"
 crossbeam = "0.8"
 enclose = "1.1"
-flume = { version = "0.10", features = ["async"] }
+flume = { version = "0.11", features = ["async"] }
 futures-lite = "1.12"
 intrusive-collections = "0.9"
 lazy_static = "1.4"

glommio/src/io/buffered_file_stream.rs

@@ -502,11 +502,7 @@ macro_rules! do_seek {
             }
             SeekFrom::End(pos) => match $source.take() {
                 None => {
-                    let source = $self
-                        .reactor
-                        .upgrade()
-                        .unwrap()
-                        .statx($fileobj.as_raw_fd(), &$fileobj.path().unwrap());
+                    let source = $self.reactor.upgrade().unwrap().statx($fileobj.as_raw_fd());
                     source.add_waiter_single($cx.waker());
                     $source = Some(source);
                     Poll::Pending

glommio/src/io/dma_file.rs

@@ -1618,6 +1618,58 @@ pub(crate) mod test {
             .expect_err("O_TMPFILE requires opening with write permissions");
     });
 
+    dma_file_test!(deleted_file_still_can_be_stat, path, _k, {
+        let file = OpenOptions::new()
+            .create_new(true)
+            .read(true)
+            .write(true)
+            .dma_open(path.join("deleted_file_still_can_be_stat"))
+            .await
+            .expect("file should open");
+        let mut buf = file.alloc_dma_buffer(512);
+        buf.as_bytes_mut().fill(2);
+        file.write_at(buf, 0)
+            .await
+            .expect("should be able to write the file");
+        file.remove().await.expect("should have removed file");
+        let stat = file
+            .stat()
+            .await
+            .expect("should be able to state unlinked but open file");
+        assert_eq!(stat.file_size, 512);
+    });
+
+    dma_file_test!(tmpfiles_have_unique_inode, path, _k, {
+        let f1 = OpenOptions::new()
+            .create_new(true)
+            .read(true)
+            .write(true)
+            .tmpfile(true)
+            .dma_open(&path)
+            .await
+            .unwrap();
+
+        let f2 = OpenOptions::new()
+            .create_new(true)
+            .read(true)
+            .write(true)
+            .tmpfile(true)
+            .dma_open(&path)
+            .await
+            .unwrap();
+
+        assert_ne!(f1.inode(), f2.inode());
+        assert_eq!(f1.stat().await.unwrap().file_size, 0);
+        assert_eq!(f2.stat().await.unwrap().file_size, 0);
+
+        let mut buf = f1.alloc_dma_buffer(512);
+        buf.as_bytes_mut().fill(2);
+        f1.write_at(buf, 0)
+            .await
+            .expect("failed to write to temporary file");
+        assert_eq!(f1.stat().await.unwrap().file_size, 512);
+    });
+
     dma_file_test!(resize_dma_buf, path, _k, {
         let file = OpenOptions::new()
             .create_new(true)

glommio/src/io/glommio_file.rs

@@ -329,13 +329,7 @@ impl GlommioFile {
 
     // Retrieve file metadata, backed by the statx(2) syscall
     pub(crate) async fn statx(&self) -> Result<Statx> {
-        let path = self.path_required("stat")?.to_owned();
-
-        let source = self
-            .reactor
-            .upgrade()
-            .unwrap()
-            .statx(self.as_raw_fd(), path.as_ref());
+        let source = self.reactor.upgrade().unwrap().statx(self.as_raw_fd());
         source.collect_rw().await.map_err(|source| {
             GlommioError::create_enhanced(
                 source,

glommio/src/reactor.rs

@@ -737,20 +737,18 @@ impl Reactor {
         source
     }
 
-    pub(crate) fn statx(&self, raw: RawFd, path: &Path) -> Source {
-        let path = CString::new(path.as_os_str().as_bytes()).expect("path contained null!");
-
+    pub(crate) fn statx(&self, raw: RawFd) -> Source {
         let statx_buf = unsafe {
             let statx_buf = mem::MaybeUninit::<Statx>::zeroed();
             statx_buf.assume_init()
         };
 
         let source = self.new_source(
             raw,
-            SourceType::Statx(path, Box::new(RefCell::new(statx_buf))),
+            SourceType::Statx(Box::new(RefCell::new(statx_buf))),
             None,
         );
-        self.sys.statx(&source);
+        self.sys.statx_fd(&source);
         source
     }
 

glommio/src/sys/source.rs

@@ -54,7 +54,7 @@ pub(crate) enum SourceType {
     Close,
     LinkRings,
     ForeignNotifier(u64, bool),
-    Statx(CString, Box<RefCell<Statx>>),
+    Statx(Box<RefCell<Statx>>),
     Timeout(TimeSpec64, u32),
     Connect(nix::sys::socket::SockaddrStorage),
     Accept(SockAddrStorage),
@@ -73,7 +73,7 @@ impl TryFrom<SourceType> for Statx {
 
     fn try_from(value: SourceType) -> Result<Self, Self::Error> {
         match value {
-            SourceType::Statx(_, buf) => Ok(buf.into_inner()),
+            SourceType::Statx(buf) => Ok(buf.into_inner()),
             src => Err(GlommioError::ReactorError(
                 ReactorErrorKind::IncorrectSourceType(format!("{src:?}")),
             )),

glommio/src/sys/uring.rs

@@ -70,7 +70,7 @@ enum UringOpDescriptor {
     LinkTimeout(*const uring_sys::__kernel_timespec),
     Accept(*mut SockAddrStorage),
     Fallocate(u64, u64, libc::c_int),
-    Statx(*const u8, *mut Statx),
+    StatxFd(RawFd, *mut Statx),
     Timeout(*const uring_sys::__kernel_timespec, u32),
     TimeoutRemove(u64),
     SockSend(*const u8, usize, i32),
@@ -334,12 +334,12 @@ fn fill_sqe<F>(
                 let flags = FallocateFlags::from_bits_truncate(flags);
                 sqe.prep_fallocate(op.fd, offset, size, flags);
             }
-            UringOpDescriptor::Statx(path, statx_buf) => {
-                let flags = StatxFlags::AT_STATX_SYNC_AS_STAT | StatxFlags::AT_NO_AUTOMOUNT;
+            UringOpDescriptor::StatxFd(fd, statx_buf) => {
+                let flags = StatxFlags::AT_STATX_SYNC_AS_STAT
+                    | StatxFlags::AT_NO_AUTOMOUNT
+                    | StatxFlags::AT_EMPTY_PATH;
                 let mode = StatxMode::from_bits_truncate(0x7ff);
-
-                let path = CStr::from_ptr(path as _);
-                sqe.prep_statx(-1, path, flags, mode, &mut *statx_buf);
+                sqe.prep_statx(fd, Default::default(), flags, mode, &mut *statx_buf);
             }
             UringOpDescriptor::Timeout(timespec, events) => {
                 sqe.prep_timeout(&*timespec, events, TimeoutFlags::empty());
@@ -1641,12 +1641,11 @@ impl Reactor {
         );
     }
 
-    pub(crate) fn statx(&self, source: &Source) {
+    pub(crate) fn statx_fd(&self, source: &Source) {
         let op = match &*source.source_type() {
-            SourceType::Statx(path, buf) => {
-                let path = path.as_c_str().as_ptr();
+            SourceType::Statx(buf) => {
                 let buf = buf.as_ptr();
-                UringOpDescriptor::Statx(path as _, buf)
+                UringOpDescriptor::StatxFd(source.raw(), buf)
             }
             _ => panic!("Unexpected source for statx operation"),
         };