Add fcntl subcommand F_GETOWN/F_SETOWN

kernel/aster-nix/src/fs/file_table.rs

@@ -12,9 +12,15 @@ use super::{
     utils::{AccessMode, InodeMode},
 };
 use crate::{
-    events::{Events, Observer, Subject},
+    events::{Events, IoEvents, Observer, Subject},
+    fs::utils::StatusFlags,
     net::socket::Socket,
     prelude::*,
+    process::{
+        process_table,
+        signal::{constants::SIGIO, signals::kernel::KernelSignal},
+        Pid, Process,
+    },
 };
 
 pub type FileDesc = i32;
@@ -232,6 +238,7 @@ pub struct FileTableEntry {
     file: Arc<dyn FileLike>,
     flags: AtomicU8,
     subject: Subject<FdEvents>,
+    owner: RwLock<Option<Owner>>,
 }
 
 impl FileTableEntry {
@@ -240,6 +247,7 @@ impl FileTableEntry {
             file,
             flags: AtomicU8::new(flags.bits()),
             subject: Subject::new(),
+            owner: RwLock::new(None),
         }
     }
 
@@ -247,6 +255,48 @@ impl FileTableEntry {
         &self.file
     }
 
+    pub fn owner(&self) -> Option<Pid> {
+        self.owner.read().as_ref().map(|(pid, _)| *pid)
+    }
+
+    /// Set a process (group) as owner of the file descriptor.
+    ///
+    /// Such that this process (group) will receive `SIGIO` and `SIGURG` signals
+    /// for I/O events on the file descriptor, if `O_ASYNC` status flag is set
+    /// on this file.
+    pub fn set_owner(&self, owner: Pid) -> Result<()> {
+        if self.owner().is_some_and(|pid| pid == owner) {
+            return Ok(());
+        }
+
+        // Unset the owner if the given pid is zero.
+        let new_owner = if owner == 0 {
+            None
+        } else {
+            let process = process_table::get_process(owner as _).ok_or(Error::with_message(
+                Errno::ESRCH,
+                "cannot set_owner with an invalid pid",
+            ))?;
+            let observer = OwnerObserver::new(self.file.clone(), Arc::downgrade(&process));
+            Some((owner, observer))
+        };
+
+        let mut self_owner = self.owner.write();
+        if let Some((_, observer)) = self_owner.as_ref() {
+            let _ = self.file.unregister_observer(&Arc::downgrade(observer));
+        }
+
+        *self_owner = match new_owner {
+            None => None,
+            Some((pid, observer)) => {
+                self.file
+                    .register_observer(observer.weak_self(), IoEvents::empty())?;
+                Some((pid, observer))
+            }
+        };
+        Ok(())
+    }
+
     pub fn flags(&self) -> FdFlags {
         FdFlags::from_bits(self.flags.load(Ordering::Relaxed)).unwrap()
     }
@@ -274,6 +324,7 @@ impl Clone for FileTableEntry {
             file: self.file.clone(),
             flags: AtomicU8::new(self.flags.load(Ordering::Relaxed)),
             subject: Subject::new(),
+            owner: RwLock::new(self.owner.read().clone()),
         }
     }
 }
@@ -284,3 +335,35 @@ bitflags! {
         const CLOEXEC = 1;
     }
 }
+
+type Owner = (Pid, Arc<dyn Observer<IoEvents>>);
+
+struct OwnerObserver {
+    file: Arc<dyn FileLike>,
+    owner: Weak<Process>,
+    weak_self: Weak<Self>,
+}
+
+impl OwnerObserver {
+    pub fn new(file: Arc<dyn FileLike>, owner: Weak<Process>) -> Arc<Self> {
+        Arc::new_cyclic(|weak_ref| Self {
+            file,
+            owner,
+            weak_self: weak_ref.clone(),
+        })
+    }
+
+    pub fn weak_self(&self) -> Weak<Self> {
+        self.weak_self.clone()
+    }
+}
+
+impl Observer<IoEvents> for OwnerObserver {
+    fn on_events(&self, events: &IoEvents) {
+        if self.file.status_flags().contains(StatusFlags::O_ASYNC)
+            && let Some(process) = self.owner.upgrade()
+        {
+            process.enqueue_signal(KernelSignal::new(SIGIO));
+        }
+    }
+}

kernel/aster-nix/src/net/socket/unix/stream/connected.rs

@@ -2,7 +2,7 @@
 
 use super::endpoint::Endpoint;
 use crate::{
-    events::IoEvents,
+    events::{IoEvents, Observer},
     net::socket::{unix::addr::UnixSocketAddrBound, SockShutdownCmd},
     prelude::*,
     process::signal::Poller,
@@ -40,4 +40,19 @@ impl Connected {
     pub(super) fn poll(&self, mask: IoEvents, poller: Option<&mut Poller>) -> IoEvents {
         self.local_endpoint.poll(mask, poller)
     }
+
+    pub(super) fn register_observer(
+        &self,
+        observer: Weak<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        self.local_endpoint.register_observer(observer, mask)
+    }
+
+    pub(super) fn unregister_observer(
+        &self,
+        observer: &Weak<dyn Observer<IoEvents>>,
+    ) -> Option<Weak<dyn Observer<IoEvents>>> {
+        self.local_endpoint.unregister_observer(observer)
+    }
 }

kernel/aster-nix/src/net/socket/unix/stream/endpoint.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: MPL-2.0
 
 use crate::{
-    events::IoEvents,
+    events::{IoEvents, Observer},
     fs::utils::{Channel, Consumer, Producer},
     net::socket::{unix::addr::UnixSocketAddrBound, SockShutdownCmd},
     prelude::*,
@@ -88,6 +88,38 @@ impl Endpoint {
 
         events
     }
+
+    pub(super) fn register_observer(
+        &self,
+        observer: Weak<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        if mask.contains(IoEvents::IN) {
+            self.reader.register_observer(observer.clone(), mask)?
+        }
+
+        if mask.contains(IoEvents::OUT) {
+            self.writer.register_observer(observer, mask)?
+        }
+
+        Ok(())
+    }
+
+    pub(super) fn unregister_observer(
+        &self,
+        observer: &Weak<dyn Observer<IoEvents>>,
+    ) -> Option<Weak<dyn Observer<IoEvents>>> {
+        let observer0 = self.reader.unregister_observer(observer);
+        let observer1 = self.writer.unregister_observer(observer);
+
+        if observer0.is_some() {
+            observer0
+        } else if observer1.is_some() {
+            observer1
+        } else {
+            None
+        }
+    }
 }
 
 const DAFAULT_BUF_SIZE: usize = 4096;

kernel/aster-nix/src/net/socket/unix/stream/init.rs

@@ -2,7 +2,7 @@
 
 use super::{connected::Connected, endpoint::Endpoint, listener::push_incoming};
 use crate::{
-    events::IoEvents,
+    events::{IoEvents, Observer},
     fs::{
         fs_resolver::{split_path, FsPath},
         path::Dentry,
@@ -66,6 +66,22 @@ impl Init {
     pub(super) fn poll(&self, mask: IoEvents, poller: Option<&mut Poller>) -> IoEvents {
         self.pollee.poll(mask, poller)
     }
+
+    pub(super) fn register_observer(
+        &self,
+        observer: Weak<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        self.pollee.register_observer(observer, mask);
+        Ok(())
+    }
+
+    pub(super) fn unregister_observer(
+        &self,
+        observer: &Weak<dyn Observer<IoEvents>>,
+    ) -> Option<Weak<dyn Observer<IoEvents>>> {
+        self.pollee.unregister_observer(observer)
+    }
 }
 
 fn create_socket_file(path: &str) -> Result<Arc<Dentry>> {

kernel/aster-nix/src/net/socket/unix/stream/listener.rs

@@ -4,7 +4,7 @@ use keyable_arc::KeyableWeak;
 
 use super::{connected::Connected, endpoint::Endpoint, UnixStreamSocket};
 use crate::{
-    events::IoEvents,
+    events::{IoEvents, Observer},
     fs::{file_handle::FileLike, path::Dentry, utils::Inode},
     net::socket::{
         unix::addr::{UnixSocketAddr, UnixSocketAddrBound},
@@ -51,6 +51,25 @@ impl Listener {
         let backlog = BACKLOG_TABLE.get_backlog(addr).unwrap();
         backlog.poll(mask, poller)
     }
+
+    pub(super) fn register_observer(
+        &self,
+        observer: Weak<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        let addr = self.addr();
+        let backlog = BACKLOG_TABLE.get_backlog(addr)?;
+        backlog.register_observer(observer, mask)
+    }
+
+    pub(super) fn unregister_observer(
+        &self,
+        observer: &Weak<dyn Observer<IoEvents>>,
+    ) -> Option<Weak<dyn Observer<IoEvents>>> {
+        let addr = self.addr();
+        let backlog = BACKLOG_TABLE.get_backlog(addr).ok()?;
+        backlog.unregister_observer(observer)
+    }
 }
 
 static BACKLOG_TABLE: BacklogTable = BacklogTable::new();
@@ -169,6 +188,22 @@ impl Backlog {
         let _lock = self.incoming_endpoints.lock();
         self.pollee.poll(mask, poller)
     }
+
+    fn register_observer(
+        &self,
+        observer: Weak<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        self.pollee.register_observer(observer, mask);
+        Ok(())
+    }
+
+    fn unregister_observer(
+        &self,
+        observer: &Weak<dyn Observer<IoEvents>>,
+    ) -> Option<Weak<dyn Observer<IoEvents>>> {
+        self.pollee.unregister_observer(observer)
+    }
 }
 
 fn create_keyable_inode(dentry: &Arc<Dentry>) -> KeyableWeak<dyn Inode> {

kernel/aster-nix/src/net/socket/unix/stream/socket.rs

@@ -11,7 +11,7 @@ use super::{
     listener::{unregister_backlog, Listener},
 };
 use crate::{
-    events::IoEvents,
+    events::{IoEvents, Observer},
     fs::{
         file_handle::FileLike,
         fs_resolver::FsPath,
@@ -166,6 +166,31 @@ impl FileLike for UnixStreamSocket {
         self.set_nonblocking(new_flags.contains(StatusFlags::O_NONBLOCK));
         Ok(())
     }
+
+    fn register_observer(
+        &self,
+        observer: Weak<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        let inner = self.state.write();
+        match &*inner {
+            State::Init(init) => init.register_observer(observer, mask),
+            State::Listen(listen) => listen.register_observer(observer, mask),
+            State::Connected(connected) => connected.register_observer(observer, mask),
+        }
+    }
+
+    fn unregister_observer(
+        &self,
+        observer: &Weak<dyn Observer<IoEvents>>,
+    ) -> Option<Weak<dyn Observer<IoEvents>>> {
+        let inner = self.state.write();
+        match &*inner {
+            State::Init(init) => init.unregister_observer(observer),
+            State::Listen(listen) => listen.unregister_observer(observer),
+            State::Connected(connected) => connected.unregister_observer(observer),
+        }
+    }
 }
 
 impl Socket for UnixStreamSocket {

kernel/aster-nix/src/syscall/fcntl.rs

@@ -7,6 +7,7 @@ use crate::{
         utils::StatusFlags,
     },
     prelude::*,
+    process::Pid,
 };
 
 pub fn sys_fcntl(fd: FileDesc, cmd: i32, arg: u64) -> Result<SyscallReturn> {
@@ -80,6 +81,27 @@ pub fn sys_fcntl(fd: FileDesc, cmd: i32, arg: u64) -> Result<SyscallReturn> {
             file.set_status_flags(new_status_flags)?;
             Ok(SyscallReturn::Return(0))
         }
+        FcntlCmd::F_SETOWN => {
+            let current = current!();
+            let file_table = current.file_table().lock();
+            let file_entry = file_table.get_entry(fd)?;
+            // A process ID is specified as a positive value; a process group ID is specified as a negative value.
+            let abs_arg = (arg as i32).unsigned_abs();
+            if abs_arg > i32::MAX as u32 {
+                return_errno_with_message!(Errno::EINVAL, "process (group) id overflowed");
+            }
+            let pid = Pid::try_from(abs_arg)
+                .map_err(|_| Error::with_message(Errno::EINVAL, "invalid process (group) id"))?;
+            file_entry.set_owner(pid)?;
+            Ok(SyscallReturn::Return(0))
+        }
+        FcntlCmd::F_GETOWN => {
+            let current = current!();
+            let file_table = current.file_table().lock();
+            let file_entry = file_table.get_entry(fd)?;
+            let pid = file_entry.owner().unwrap_or(0);
+            Ok(SyscallReturn::Return(pid as _))
+        }
     }
 }
 
@@ -92,5 +114,7 @@ enum FcntlCmd {
     F_SETFD = 2,
     F_GETFL = 3,
     F_SETFL = 4,
+    F_SETOWN = 8,
+    F_GETOWN = 9,
     F_DUPFD_CLOEXEC = 1030,
 }

kernel/aster-nix/src/syscall/socket.rs

@@ -23,7 +23,8 @@ pub fn sys_socket(domain: i32, type_: i32, protocol: i32) -> Result<SyscallRetur
     );
     let nonblocking = sock_flags.contains(SockFlags::SOCK_NONBLOCK);
     let file_like = match (domain, sock_type, protocol) {
-        (CSocketAddrFamily::AF_UNIX, SockType::SOCK_STREAM, _) => {
+        // FIXME: SOCK_SEQPACKET is added to run fcntl_test, not supported yet.
+        (CSocketAddrFamily::AF_UNIX, SockType::SOCK_STREAM | SockType::SOCK_SEQPACKET, _) => {
             UnixStreamSocket::new(nonblocking) as Arc<dyn FileLike>
         }
         (