refactor: define detailed error types

Cargo.toml

@@ -14,25 +14,30 @@ pg16 = ["pgrx/pg16", "pgrx-tests/pg16"]
 pg_test = []
 
 [dependencies]
+arrayvec.workspace = true
+bincode.workspace = true
+bytemuck.workspace = true
+byteorder.workspace = true
+half.workspace = true
 libc.workspace = true
 log.workspace = true
+memmap2.workspace = true
+num-traits.workspace = true
+paste.workspace = true
+rand.workspace = true
+rustix.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-validator.workspace = true
-rustix.workspace = true
 thiserror.workspace = true
-byteorder.workspace = true
-bincode.workspace = true
-half.workspace = true
-num-traits.workspace = true
-rand.workspace = true
-bytemuck.workspace = true
-service = { path = "crates/service" }
+validator.workspace = true
 detect = { path = "crates/detect" }
+send_fd = { path = "crates/send_fd" }
+service = { path = "crates/service" }
+interprocess_atomic_wait = { path = "crates/interprocess-atomic-wait" }
+memfd = { path = "crates/memfd" }
 pgrx = { version = "0.11.3", default-features = false, features = [] }
-env_logger = "0.10.0"
-toml = "0.8.8"
-arrayvec = "0.7.4"
+env_logger = "0.11.2"
+toml = "0.8.10"
 
 [dev-dependencies]
 pgrx-tests = "0.11.3"
@@ -58,24 +63,29 @@ version = "0.0.0"
 edition = "2021"
 
 [workspace.dependencies]
-libc = "~0.2"
-log = "~0.4"
-serde = "~1.0"
-serde_json = "1"
-thiserror = "~1.0"
+arrayvec = "~0.7"
 bincode = "~1.3"
-byteorder = "~1.5"
 bytemuck = { version = "~1.14", features = ["extern_crate_alloc"] }
+byteorder = "~1.5"
 half = { version = "~2.3", features = [
     "bytemuck",
     "num-traits",
     "serde",
     "use-intrinsics",
+    "rand_distr",
 ] }
+libc = "~0.2"
+log = "~0.4"
+memmap2 = "0.9.4"
 num-traits = "~0.2"
-validator = { version = "~0.16", features = ["derive"] }
+paste = "~1.0"
+rand = "0.8.5"
 rustix = { version = "~0.38", features = ["fs", "net", "mm"] }
-rand = "~0.8"
+serde = "~1.0"
+serde_json = "~1.0"
+thiserror = "~1.0"
+uuid = { version = "1.7.0", features = ["v4", "serde"] }
+validator = { version = "~0.16", features = ["derive"] }
 
 [profile.dev]
 panic = "unwind"

crates/c/Cargo.toml

@@ -4,9 +4,9 @@ version.workspace = true
 edition.workspace = true
 
 [dev-dependencies]
-half = { version = "~2.3", features = ["use-intrinsics", "rand_distr"] }
+half.workspace = true
+rand.workspace = true
 detect = { path = "../detect" }
-rand = "0.8.5"
 
 [build-dependencies]
 cc = "1.0"

crates/detect/Cargo.toml

@@ -4,5 +4,5 @@ version.workspace = true
 edition.workspace = true
 
 [dependencies]
-std_detect = { git = "https://github.com/tensorchord/stdarch.git", branch = "avx512fp16" }
 rustix.workspace = true
+std_detect = { git = "https://github.com/tensorchord/stdarch.git", branch = "avx512fp16" }

crates/interprocess-atomic-wait/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "interprocess_atomic_wait"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+libc.workspace = true
+
+[target.'cfg(target_os = "macos")'.dependencies]
+ulock-sys = "0.1.0"
+
+[lints]
+rust.internal_features = "allow"
+rust.unsafe_op_in_unsafe_fn = "forbid"
+rust.unused_lifetimes = "warn"
+rust.unused_qualifications = "warn"

crates/interprocess-atomic-wait/src/lib.rs

@@ -0,0 +1,91 @@
+use std::sync::atomic::AtomicU32;
+use std::time::Duration;
+
+#[cfg(target_os = "linux")]
+#[inline(always)]
+pub fn wait(futex: &AtomicU32, value: u32, timeout: Duration) {
+    let timeout = libc::timespec {
+        tv_sec: i64::try_from(timeout.as_secs()).expect("Timeout is overflow."),
+        tv_nsec: timeout.subsec_nanos().into(),
+    };
+    unsafe {
+        libc::syscall(
+            libc::SYS_futex,
+            futex.as_ptr(),
+            libc::FUTEX_WAIT,
+            value,
+            &timeout,
+        );
+    }
+}
+
+#[cfg(target_os = "linux")]
+#[inline(always)]
+pub fn wake(futex: &AtomicU32) {
+    unsafe {
+        libc::syscall(libc::SYS_futex, futex.as_ptr(), libc::FUTEX_WAKE, i32::MAX);
+    }
+}
+
+#[cfg(target_os = "macos")]
+#[inline(always)]
+pub fn wait(futex: &AtomicU32, value: u32, timeout: Duration) {
+    let timeout = u32::try_from(timeout.as_millis()).expect("Timeout is overflow.");
+    unsafe {
+        // https://github.com/apple-oss-distributions/xnu/blob/main/bsd/kern/sys_ulock.c#L531
+        ulock_sys::__ulock_wait(
+            ulock_sys::darwin19::UL_COMPARE_AND_WAIT_SHARED,
+            futex.as_ptr().cast(),
+            value as _,
+            timeout,
+        );
+    }
+}
+
+#[cfg(target_os = "macos")]
+#[inline(always)]
+pub fn wake(futex: &AtomicU32) {
+    unsafe {
+        ulock_sys::__ulock_wake(
+            ulock_sys::darwin19::UL_COMPARE_AND_WAIT_SHARED,
+            futex.as_ptr().cast(),
+            0,
+        );
+    }
+}
+
+#[cfg(target_os = "freebsd")]
+#[inline(always)]
+pub fn wait(futex: &AtomicU32, value: u32, timeout: Duration) {
+    let ptr: *const AtomicU32 = futex;
+    let mut timeout = libc::timespec {
+        tv_sec: i64::try_from(timeout.as_secs()).expect("Timeout is overflow."),
+        tv_nsec: timeout.subsec_nanos().into(),
+    };
+    unsafe {
+        // https://github.com/freebsd/freebsd-src/blob/main/sys/kern/kern_umtx.c#L3943
+        // https://github.com/freebsd/freebsd-src/blob/main/sys/kern/kern_umtx.c#L3836
+        libc::_umtx_op(
+            ptr as *mut libc::c_void,
+            libc::UMTX_OP_WAIT_UINT,
+            value as libc::c_ulong,
+            std::mem::size_of_val(&timeout) as *mut std::ffi::c_void,
+            std::ptr::addr_of_mut!(timeout).cast(),
+        );
+    };
+}
+
+#[cfg(target_os = "freebsd")]
+#[inline(always)]
+pub fn wake(futex: &AtomicU32) {
+    let ptr: *const AtomicU32 = futex;
+    unsafe {
+        libc::_umtx_op(
+            ptr as *mut libc::c_void,
+            libc::UMTX_OP_WAKE,
+            i32::MAX as libc::c_ulong,
+            core::ptr::null_mut(),
+            core::ptr::null_mut(),
+        );
+    };
+}

crates/memfd/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "memfd"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+rand.workspace = true
+rustix.workspace = true
+detect = { path = "../detect" }
+
+[lints]
+rust.internal_features = "allow"
+rust.unsafe_op_in_unsafe_fn = "forbid"
+rust.unused_lifetimes = "warn"
+rust.unused_qualifications = "warn"

crates/memfd/src/lib.rs

@@ -0,0 +1,70 @@
+use std::os::fd::OwnedFd;
+
+#[cfg(target_os = "linux")]
+pub fn memfd_create() -> std::io::Result<OwnedFd> {
+    if detect::linux::detect_memfd() {
+        use rustix::fs::MemfdFlags;
+        Ok(rustix::fs::memfd_create(
+            format!(".memfd.MEMFD.{:x}", std::process::id()),
+            MemfdFlags::empty(),
+        )?)
+    } else {
+        use rustix::fs::Mode;
+        use rustix::fs::OFlags;
+        // POSIX fcntl locking do not support shmem, so we use a regular file here.
+        // reference: https://man7.org/linux/man-pages/man3/fcntl.3p.html
+        // However, Linux shmem supports fcntl locking.
+        let name = format!(
+            ".shm.MEMFD.{:x}.{:x}",
+            std::process::id(),
+            rand::random::<u32>()
+        );
+        let fd = rustix::fs::open(
+            &name,
+            OFlags::RDWR | OFlags::CREATE | OFlags::EXCL,
+            Mode::RUSR | Mode::WUSR,
+        )?;
+        rustix::fs::unlink(&name)?;
+        Ok(fd)
+    }
+}
+
+#[cfg(target_os = "macos")]
+pub fn memfd_create() -> std::io::Result<OwnedFd> {
+    use rustix::fs::Mode;
+    use rustix::fs::OFlags;
+    // POSIX fcntl locking do not support shmem, so we use a regular file here.
+    // reference: https://man7.org/linux/man-pages/man3/fcntl.3p.html
+    let name = format!(
+        ".shm.MEMFD.{:x}.{:x}",
+        std::process::id(),
+        rand::random::<u32>()
+    );
+    let fd = rustix::fs::open(
+        &name,
+        OFlags::RDWR | OFlags::CREATE | OFlags::EXCL,
+        Mode::RUSR | Mode::WUSR,
+    )?;
+    rustix::fs::unlink(&name)?;
+    Ok(fd)
+}
+
+#[cfg(target_os = "freebsd")]
+pub fn memfd_create() -> std::io::Result<OwnedFd> {
+    use rustix::fs::Mode;
+    use rustix::fs::OFlags;
+    // POSIX fcntl locking do not support shmem, so we use a regular file here.
+    // reference: https://man7.org/linux/man-pages/man3/fcntl.3p.html
+    let name = format!(
+        ".shm.MEMFD.{:x}.{:x}",
+        std::process::id(),
+        rand::random::<u32>()
+    );
+    let fd = rustix::fs::open(
+        &name,
+        OFlags::RDWR | OFlags::CREATE | OFlags::EXCL,
+        Mode::RUSR | Mode::WUSR,
+    )?;
+    rustix::fs::unlink(&name)?;
+    Ok(fd)
+}

crates/send_fd/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "send_fd"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+libc.workspace = true
+log.workspace = true
+rustix.workspace = true
+
+[lints]
+rust.internal_features = "allow"
+rust.unsafe_op_in_unsafe_fn = "forbid"
+rust.unused_lifetimes = "warn"
+rust.unused_qualifications = "warn"

src/utils/file_socket.rs → crates/send_fd/src/lib.rs

@@ -6,12 +6,12 @@ use std::io::{IoSlice, IoSliceMut};
 use std::os::unix::net::UnixStream;
 
 #[repr(C)]
-pub struct FileSocket {
+pub struct SendFd {
     tx: OwnedFd,
     rx: OwnedFd,
 }
 
-impl FileSocket {
+impl SendFd {
     pub fn new() -> std::io::Result<Self> {
         let (tx, rx) = UnixStream::pair()?;
         Ok(Self {
@@ -47,7 +47,7 @@ fn recv_fd(rx: BorrowedFd<'_>) -> std::io::Result<OwnedFd> {
         let mut control = RecvAncillaryBuffer::new(&mut buffer.0);
         let mut buffer_ios = [b'.'];
         let ios = IoSliceMut::new(&mut buffer_ios);
-        let returned = rustix::net::recvmsg(rx, &mut [ios], &mut control, RecvFlags::CMSG_CLOEXEC)?;
+        let returned = rustix::net::recvmsg(rx, &mut [ios], &mut control, RecvFlags::empty())?;
         if returned.flags.bits() & libc::MSG_CTRUNC as u32 != 0 {
             log::warn!("Ancillary is truncated.");
         }

crates/service/Cargo.toml

@@ -4,36 +4,31 @@ version.workspace = true
 edition.workspace = true
 
 [dependencies]
+bincode.workspace = true
+bytemuck.workspace = true
+byteorder.workspace = true
+half.workspace = true
 libc.workspace = true
 log.workspace = true
+memmap2.workspace = true
+num-traits.workspace = true
+rand.workspace = true
+rustix.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-validator.workspace = true
-rustix.workspace = true
 thiserror.workspace = true
-byteorder.workspace = true
-bincode.workspace = true
-half.workspace = true
-num-traits.workspace = true
-rand.workspace = true
-bytemuck.workspace = true
+uuid.workspace = true
+validator.workspace = true
 c = { path = "../c" }
 detect = { path = "../detect" }
-crc32fast = "1.3.2"
-crossbeam = "0.8.2"
-dashmap = "5.4.0"
+crc32fast = "1.4.0"
+crossbeam = "0.8.4"
+dashmap = "5.5.3"
 parking_lot = "0.12.1"
-memoffset = "0.9.0"
-arrayvec = { version = "0.7.3", features = ["serde"] }
-memmap2 = "0.9.0"
-rayon = "1.6.1"
-uuid = { version = "1.6.1", features = ["v4", "serde"] }
+rayon = "1.8.1"
 arc-swap = "1.6.0"
 multiversion = "0.7.3"
 
-[target.'cfg(target_os = "macos")'.dependencies]
-ulock-sys = "0.1.0"
-
 [lints]
 clippy.derivable_impls = "allow"
 clippy.len_without_is_empty = "allow"