Add support for epoll

services/libs/jinux-std/src/events/subject.rs

@@ -21,9 +21,9 @@ impl<E: Events> Subject<E> {
     }
 
     /// Unregister an observer.
-    pub fn unregister_observer(&self, observer: Weak<dyn Observer<E>>) {
+    pub fn unregister_observer(&self, observer: &Weak<dyn Observer<E>>) {
         let mut observers = self.observers.lock();
-        observers.retain(|e| !Weak::ptr_eq(&e, &observer));
+        observers.retain(|e| !Weak::ptr_eq(&e, observer));
     }
 
     /// Notify events to all registered observers.

services/libs/jinux-std/src/fs/epoll/epoll_file.rs

@@ -0,0 +1,485 @@
+use crate::events::Observer;
+use crate::fs::file_handle::FileLike;
+use crate::fs::file_table::{FdEvents, FileDescripter};
+use crate::fs::utils::{IoEvents, IoctlCmd, Pollee, Poller};
+
+use core::sync::atomic::{AtomicBool, Ordering};
+use core::time::Duration;
+
+use super::*;
+
+/// A file-like object that provides epoll API.
+///
+/// Conceptually, we maintain two lists: one consists of all interesting files,
+/// which can be managed by the epoll ctl commands; the other are for ready files,
+/// which are files that have some events. A epoll wait only needs to iterate the
+/// ready list and poll each file to see if the file is ready for the interesting
+/// I/O.
+///
+/// To maintain the ready list, we need to monitor interesting events that happen
+/// on the files. To do so, the `EpollFile` registers itself as an `Observer` to
+/// the monotored files. Thus, we can add a file to the ready list when an interesting
+/// event happens on the file.
+pub struct EpollFile {
+    // All interesting entries.
+    interest: Mutex<BTreeMap<FileDescripter, Arc<EpollEntry>>>,
+    // Entries that are probably ready (having events happened).
+    ready: Mutex<VecDeque<Arc<EpollEntry>>>,
+    // EpollFile itself is also pollable
+    pollee: Pollee,
+    // Any EpollFile is wrapped with Arc when created.
+    weak_self: Weak<Self>,
+}
+
+impl EpollFile {
+    /// Creates a new epoll file.
+    pub fn new() -> Arc<Self> {
+        Arc::new_cyclic(|me| Self {
+            interest: Mutex::new(BTreeMap::new()),
+            ready: Mutex::new(VecDeque::new()),
+            pollee: Pollee::new(IoEvents::empty()),
+            weak_self: me.clone(),
+        })
+    }
+
+    /// Control the interest list of the epoll file.
+    pub fn control(&self, cmd: &EpollCtl) -> Result<()> {
+        match *cmd {
+            EpollCtl::Add(fd, ep_event, ep_flags) => self.add_interest(fd, ep_event, ep_flags),
+            EpollCtl::Del(fd) => {
+                self.del_interest(fd)?;
+                self.unregister_from_file_table_entry(fd);
+                Ok(())
+            }
+            EpollCtl::Mod(fd, ep_event, ep_flags) => self.mod_interest(fd, ep_event, ep_flags),
+        }
+    }
+
+    fn add_interest(
+        &self,
+        fd: FileDescripter,
+        ep_event: EpollEvent,
+        ep_flags: EpollFlags,
+    ) -> Result<()> {
+        self.warn_unsupported_flags(&ep_flags);
+
+        let current = current!();
+        let file_table = current.file_table().lock();
+        let file_table_entry = file_table.get_entry(fd)?;
+        let file = file_table_entry.file();
+        let weak_file = Arc::downgrade(file);
+        let mask = ep_event.events;
+        let entry = EpollEntry::new(fd, weak_file, ep_event, ep_flags, self.weak_self.clone());
+
+        // Add the new entry to the interest list and start monitering its events
+        let mut interest = self.interest.lock();
+        if interest.contains_key(&fd) {
+            return_errno_with_message!(Errno::EEXIST, "the fd has been added");
+        }
+        file.register_observer(entry.clone(), IoEvents::all())?;
+        interest.insert(fd, entry.clone());
+        // Register self to the file table entry
+        file_table_entry.register_observer(self.weak_self.clone() as _);
+        let file = file.clone();
+        drop(file_table);
+        drop(interest);
+
+        // Add the new entry to the ready list if the file is ready
+        let events = file.poll(mask, None);
+        if !events.is_empty() {
+            self.push_ready(entry);
+        }
+        Ok(())
+    }
+
+    fn del_interest(&self, fd: FileDescripter) -> Result<()> {
+        let mut interest = self.interest.lock();
+        let entry = interest
+            .remove(&fd)
+            .ok_or_else(|| Error::with_message(Errno::ENOENT, "fd is not in the interest list"))?;
+
+        // If this epoll entry is in the ready list, then we should delete it.
+        // But unfortunately, deleting an entry from the ready list has a
+        // complexity of O(N).
+        //
+        // To optimize the performance, we only mark the epoll entry as
+        // deleted at this moment. The real deletion happens when the ready list
+        // is scanned in EpolFile::wait.
+        entry.set_deleted();
+
+        let file = match entry.file() {
+            Some(file) => file,
+            // TODO: should we warn about it?
+            None => return Ok(()),
+        };
+
+        file.unregister_observer(&(entry as _)).unwrap();
+        Ok(())
+    }
+
+    fn mod_interest(
+        &self,
+        fd: FileDescripter,
+        new_ep_event: EpollEvent,
+        new_ep_flags: EpollFlags,
+    ) -> Result<()> {
+        self.warn_unsupported_flags(&new_ep_flags);
+
+        // Update the epoll entry
+        let interest = self.interest.lock();
+        let entry = interest
+            .get(&fd)
+            .ok_or_else(|| Error::with_message(Errno::ENOENT, "fd is not in the interest list"))?;
+        if entry.is_deleted() {
+            return_errno_with_message!(Errno::ENOENT, "fd is not in the interest list");
+        }
+        let new_mask = new_ep_event.events;
+        entry.update(new_ep_event, new_ep_flags);
+        let entry = entry.clone();
+        drop(interest);
+
+        // Add the updated entry to the ready list if the file is ready
+        let file = match entry.file() {
+            Some(file) => file,
+            None => return Ok(()),
+        };
+        let events = file.poll(new_mask, None);
+        if !events.is_empty() {
+            self.push_ready(entry);
+        }
+        Ok(())
+    }
+
+    fn unregister_from_file_table_entry(&self, fd: FileDescripter) {
+        let current = current!();
+        let file_table = current.file_table().lock();
+        if let Ok(entry) = file_table.get_entry(fd) {
+            entry.unregister_observer(&(self.weak_self.clone() as _));
+        }
+    }
+
+    /// Wait for interesting events happen on the files in the interest list
+    /// of the epoll file.
+    ///
+    /// This method blocks until either some interesting events happen or
+    /// the timeout expires or a signal arrives. The first case returns
+    /// `Ok(events)`, where `events` is a `Vec` containing at most `max_events`
+    /// number of `EpollEvent`s. The second and third case returns errors.
+    ///
+    /// When `max_events` equals to zero, the method returns when the timeout
+    /// expires or a signal arrives.
+    pub fn wait(&self, max_events: usize, timeout: Option<&Duration>) -> Result<Vec<EpollEvent>> {
+        let mut ep_events = Vec::new();
+        let mut poller = None;
+        loop {
+            // Try to pop some ready entries
+            if self.pop_ready(max_events, &mut ep_events) > 0 {
+                return Ok(ep_events);
+            }
+
+            // Return immediately if specifying a timeout of zero
+            if timeout.is_some() && timeout.as_ref().unwrap().is_zero() {
+                return Ok(ep_events);
+            }
+
+            // If no ready entries for now, wait for them
+            if poller.is_none() {
+                poller = Some(Poller::new());
+                let events = self.pollee.poll(IoEvents::IN, poller.as_ref());
+                if !events.is_empty() {
+                    continue;
+                }
+            }
+
+            // FIXME: respect timeout parameter
+            poller.as_ref().unwrap().wait();
+        }
+    }
+
+    fn push_ready(&self, entry: Arc<EpollEntry>) {
+        let mut ready = self.ready.lock();
+        if entry.is_deleted() {
+            return;
+        }
+
+        if !entry.is_ready() {
+            entry.set_ready();
+            ready.push_back(entry);
+        }
+
+        // Even if the entry is already set to ready, there might be new events that we are interested in.
+        // Wake the poller anyway.
+        self.pollee.add_events(IoEvents::IN);
+    }
+
+    fn pop_ready(&self, max_events: usize, ep_events: &mut Vec<EpollEvent>) -> usize {
+        let mut count_events = 0;
+        let mut ready = self.ready.lock();
+        let mut pop_quota = ready.len();
+        loop {
+            // Pop some ready entries per round.
+            //
+            // Since the popped ready entries may contain "false positive" and
+            // we want to return as many results as possible, this has to
+            // be done in a loop.
+            let pop_count = (max_events - count_events).min(pop_quota);
+            if pop_count == 0 {
+                break;
+            }
+            let ready_entries: Vec<Arc<EpollEntry>> = ready
+                .drain(..pop_count)
+                .filter(|entry| !entry.is_deleted())
+                .collect();
+            pop_quota -= pop_count;
+
+            // Examine these ready entries, which are candidates for the results
+            // to be returned.
+            for entry in ready_entries {
+                let (ep_event, ep_flags) = entry.event_and_flags();
+                // If this entry's file is ready, save it in the output array.
+                // EPOLLHUP and EPOLLERR should always be reported.
+                let ready_events = entry.poll() & (ep_event.events | IoEvents::HUP | IoEvents::ERR);
+                if !ready_events.is_empty() {
+                    ep_events.push(EpollEvent::new(ready_events, ep_event.user_data));
+                    count_events += 1;
+                }
+
+                // If the epoll entry is neither edge-triggered or one-shot, then we should
+                // keep the entry in the ready list.
+                if !ep_flags.intersects(EpollFlags::ONE_SHOT | EpollFlags::EDGE_TRIGGER) {
+                    ready.push_back(entry);
+                }
+                // Otherwise, the entry is indeed removed the ready list and we should reset
+                // its ready flag.
+                else {
+                    entry.reset_ready();
+                    // For EPOLLONESHOT flag, this entry should also be removed from the interest list
+                    if ep_flags.intersects(EpollFlags::ONE_SHOT) {
+                        self.del_interest(entry.fd())
+                            .expect("this entry should be in the interest list");
+                    }
+                }
+            }
+        }
+
+        // Clear the epoll file's events if no ready entries
+        if ready.len() == 0 {
+            self.pollee.del_events(IoEvents::IN);
+        }
+        count_events
+    }
+
+    fn warn_unsupported_flags(&self, flags: &EpollFlags) {
+        if flags.intersects(EpollFlags::EXCLUSIVE | EpollFlags::WAKE_UP) {
+            warn!("{:?} contains unsupported flags", flags);
+        }
+    }
+}
+
+impl Observer<FdEvents> for EpollFile {
+    fn on_events(&self, events: &FdEvents) {
+        // Delete the file from the interest list if it is closed.
+        if let FdEvents::Close(fd) = events {
+            let _ = self.del_interest(*fd);
+        }
+    }
+}
+
+impl Drop for EpollFile {
+    fn drop(&mut self) {
+        trace!("EpollFile Drop");
+        let mut interest = self.interest.lock();
+        let fds: Vec<_> = interest
+            .drain_filter(|_, _| true)
+            .map(|(fd, entry)| {
+                entry.set_deleted();
+                if let Some(file) = entry.file() {
+                    let _ = file.unregister_observer(&(entry as _));
+                }
+                fd
+            })
+            .collect();
+        drop(interest);
+
+        fds.iter()
+            .for_each(|&fd| self.unregister_from_file_table_entry(fd));
+    }
+}
+
+// Implement the common methods required by FileHandle
+impl FileLike for EpollFile {
+    fn read(&self, buf: &mut [u8]) -> Result<usize> {
+        return_errno_with_message!(Errno::EINVAL, "epoll files do not support read");
+    }
+
+    fn write(&self, buf: &[u8]) -> Result<usize> {
+        return_errno_with_message!(Errno::EINVAL, "epoll files do not support write");
+    }
+
+    fn ioctl(&self, _cmd: IoctlCmd, _arg: usize) -> Result<i32> {
+        return_errno_with_message!(Errno::EINVAL, "epoll files do not support ioctl");
+    }
+
+    fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.pollee.poll(mask, poller)
+    }
+
+    fn register_observer(
+        &self,
+        observer: Arc<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        self.pollee.register_observer(observer, mask);
+        Ok(())
+    }
+
+    fn unregister_observer(
+        &self,
+        observer: &Arc<dyn Observer<IoEvents>>,
+    ) -> Result<Arc<dyn Observer<IoEvents>>> {
+        self.pollee
+            .unregister_observer(observer)
+            .ok_or_else(|| Error::with_message(Errno::ENOENT, "observer is not registered"))
+    }
+
+    fn as_any_ref(&self) -> &dyn Any {
+        self
+    }
+}
+
+/// An epoll entry contained in an epoll file. Each epoll entry is added, modified,
+/// or deleted by the `EpollCtl` command.
+pub struct EpollEntry {
+    fd: FileDescripter,
+    file: Weak<dyn FileLike>,
+    inner: Mutex<Inner>,
+    // Whether the entry is in the ready list
+    is_ready: AtomicBool,
+    // Whether the entry has been deleted from the interest list
+    is_deleted: AtomicBool,
+    // Refers to the epoll file containing this epoll entry
+    weak_epoll: Weak<EpollFile>,
+    // An EpollEntry is always contained inside Arc
+    weak_self: Weak<Self>,
+}
+
+struct Inner {
+    event: EpollEvent,
+    flags: EpollFlags,
+}
+
+impl EpollEntry {
+    /// Creates a new epoll entry associated with the given epoll file.
+    pub fn new(
+        fd: FileDescripter,
+        file: Weak<dyn FileLike>,
+        event: EpollEvent,
+        flags: EpollFlags,
+        weak_epoll: Weak<EpollFile>,
+    ) -> Arc<Self> {
+        Arc::new_cyclic(|me| Self {
+            fd,
+            file,
+            inner: Mutex::new(Inner { event, flags }),
+            is_ready: AtomicBool::new(false),
+            is_deleted: AtomicBool::new(false),
+            weak_epoll,
+            weak_self: me.clone(),
+        })
+    }
+
+    /// Get the epoll file associated with this epoll entry.
+    pub fn epoll_file(&self) -> Option<Arc<EpollFile>> {
+        self.weak_epoll.upgrade()
+    }
+
+    /// Get an instance of `Arc` that refers to this epoll entry.
+    pub fn self_arc(&self) -> Arc<Self> {
+        self.weak_self.upgrade().unwrap()
+    }
+
+    /// Get the file associated with this epoll entry.
+    ///
+    /// Since an epoll entry only holds a weak reference to the file,
+    /// it is possible (albeit unlikely) that the file has been dropped.
+    pub fn file(&self) -> Option<Arc<dyn FileLike>> {
+        self.file.upgrade()
+    }
+
+    /// Get the epoll event associated with the epoll entry.
+    pub fn event(&self) -> EpollEvent {
+        let inner = self.inner.lock();
+        inner.event
+    }
+
+    /// Get the epoll flags associated with the epoll entry.
+    pub fn flags(&self) -> EpollFlags {
+        let inner = self.inner.lock();
+        inner.flags
+    }
+
+    /// Get the epoll event and flags that are associated with this epoll entry.
+    pub fn event_and_flags(&self) -> (EpollEvent, EpollFlags) {
+        let inner = self.inner.lock();
+        (inner.event, inner.flags)
+    }
+
+    /// Poll the events of the file associated with this epoll entry.
+    ///
+    /// If the returned events is not empty, then the file is considered ready.
+    pub fn poll(&self) -> IoEvents {
+        match self.file.upgrade() {
+            Some(file) => file.poll(IoEvents::all(), None),
+            None => IoEvents::empty(),
+        }
+    }
+
+    /// Update the epoll entry, most likely to be triggered via `EpollCtl::Mod`.
+    pub fn update(&self, event: EpollEvent, flags: EpollFlags) {
+        let mut inner = self.inner.lock();
+        *inner = Inner { event, flags }
+    }
+
+    /// Returns whether the epoll entry is in the ready list.
+    pub fn is_ready(&self) -> bool {
+        self.is_ready.load(Ordering::Relaxed)
+    }
+
+    /// Mark the epoll entry as being in the ready list.
+    pub fn set_ready(&self) {
+        self.is_ready.store(true, Ordering::Relaxed);
+    }
+
+    /// Mark the epoll entry as not being in the ready list.
+    pub fn reset_ready(&self) {
+        self.is_ready.store(false, Ordering::Relaxed)
+    }
+
+    /// Returns whether the epoll entry has been deleted from the interest list.
+    pub fn is_deleted(&self) -> bool {
+        self.is_deleted.load(Ordering::Relaxed)
+    }
+
+    /// Mark the epoll entry as having been deleted from the interest list.
+    pub fn set_deleted(&self) {
+        self.is_deleted.store(true, Ordering::Relaxed);
+    }
+
+    /// Get the file descriptor associated with the epoll entry.
+    pub fn fd(&self) -> FileDescripter {
+        self.fd
+    }
+}
+
+impl Observer<IoEvents> for EpollEntry {
+    fn on_events(&self, _events: &IoEvents) {
+        // Fast path
+        if self.is_deleted() {
+            return;
+        }
+
+        if let Some(epoll_file) = self.epoll_file() {
+            epoll_file.push_ready(self.self_arc());
+        }
+    }
+}

services/libs/jinux-std/src/fs/epoll/mod.rs

@@ -0,0 +1,72 @@
+use super::file_table::FileDescripter;
+use super::utils::IoEvents;
+use crate::prelude::*;
+
+mod epoll_file;
+
+pub use self::epoll_file::EpollFile;
+
+/// An epoll control command.
+#[derive(Copy, Clone, Debug, PartialEq)]
+pub enum EpollCtl {
+    Add(FileDescripter, EpollEvent, EpollFlags),
+    Del(FileDescripter),
+    Mod(FileDescripter, EpollEvent, EpollFlags),
+}
+
+bitflags! {
+    /// Linux's epoll flags.
+    pub struct EpollFlags: u32 {
+        const EXCLUSIVE      = (1 << 28);
+        const WAKE_UP        = (1 << 29);
+        const ONE_SHOT       = (1 << 30);
+        const EDGE_TRIGGER   = (1 << 31);
+    }
+}
+
+/// An epoll event.
+///
+/// This could be used as either an input of epoll ctl or an output of epoll wait.
+/// The memory layout is compatible with that of C's struct epoll_event.
+#[derive(Copy, Clone, Debug, PartialEq)]
+pub struct EpollEvent {
+    /// I/O events.
+    ///
+    /// When `EpollEvent` is used as inputs, this is treated as a mask of events.
+    /// When `EpollEvent` is used as outputs, this is the active events.
+    pub events: IoEvents,
+    /// A 64-bit, user-given data.
+    pub user_data: u64,
+}
+
+impl EpollEvent {
+    /// Create a new epoll event.
+    pub fn new(events: IoEvents, user_data: u64) -> Self {
+        Self { events, user_data }
+    }
+}
+
+impl From<&c_epoll_event> for EpollEvent {
+    fn from(c_event: &c_epoll_event) -> Self {
+        Self {
+            events: IoEvents::from_bits_truncate(c_event.events as u32),
+            user_data: c_event.data,
+        }
+    }
+}
+
+impl From<&EpollEvent> for c_epoll_event {
+    fn from(ep_event: &EpollEvent) -> Self {
+        Self {
+            events: ep_event.events.bits() as u32,
+            data: ep_event.user_data,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, Pod)]
+#[repr(C)]
+pub struct c_epoll_event {
+    pub events: u32,
+    pub data: u64,
+}

services/libs/jinux-std/src/fs/file_handle.rs

@@ -0,0 +1,74 @@
+//! Opend File Handle
+
+use crate::events::Observer;
+use crate::fs::utils::{IoEvents, IoctlCmd, Metadata, Poller, SeekFrom};
+use crate::prelude::*;
+use crate::tty::get_n_tty;
+
+use core::any::Any;
+
+/// The basic operations defined on a file
+pub trait FileLike: Send + Sync + Any {
+    fn read(&self, buf: &mut [u8]) -> Result<usize> {
+        return_errno_with_message!(Errno::EINVAL, "read is not supported");
+    }
+
+    fn write(&self, buf: &[u8]) -> Result<usize> {
+        return_errno_with_message!(Errno::EINVAL, "write is not supported");
+    }
+
+    fn ioctl(&self, cmd: IoctlCmd, arg: usize) -> Result<i32> {
+        match cmd {
+            IoctlCmd::TCGETS => {
+                // FIXME: only a work around
+                let tty = get_n_tty();
+                tty.ioctl(cmd, arg)
+            }
+            _ => panic!("Ioctl unsupported"),
+        }
+    }
+
+    fn poll(&self, _mask: IoEvents, _poller: Option<&Poller>) -> IoEvents {
+        IoEvents::empty()
+    }
+
+    fn flush(&self) -> Result<()> {
+        Ok(())
+    }
+
+    fn metadata(&self) -> Metadata {
+        panic!("metadata unsupported");
+    }
+
+    fn seek(&self, seek_from: SeekFrom) -> Result<usize> {
+        return_errno_with_message!(Errno::EINVAL, "seek is not supported");
+    }
+
+    fn clean_for_close(&self) -> Result<()> {
+        self.flush()?;
+        Ok(())
+    }
+
+    fn register_observer(
+        &self,
+        observer: Arc<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        return_errno_with_message!(Errno::EINVAL, "register_observer is not supported")
+    }
+
+    fn unregister_observer(
+        &self,
+        observer: &Arc<dyn Observer<IoEvents>>,
+    ) -> Result<Arc<dyn Observer<IoEvents>>> {
+        return_errno_with_message!(Errno::EINVAL, "unregister_observer is not supported")
+    }
+
+    fn as_any_ref(&self) -> &dyn Any;
+}
+
+impl dyn FileLike {
+    pub fn downcast_ref<T: FileLike>(&self) -> Option<&T> {
+        self.as_any_ref().downcast_ref::<T>()
+    }
+}

services/libs/jinux-std/src/fs/file_handle/file.rs

@@ -1,43 +0,0 @@
-use crate::fs::utils::{IoEvents, IoctlCmd, Metadata, Poller, SeekFrom};
-use crate::prelude::*;
-use crate::tty::get_n_tty;
-
-use core::any::Any;
-
-/// The basic operations defined on a file
-pub trait File: Send + Sync + Any {
-    fn read(&self, buf: &mut [u8]) -> Result<usize> {
-        panic!("read unsupported");
-    }
-
-    fn write(&self, buf: &[u8]) -> Result<usize> {
-        panic!("write unsupported");
-    }
-
-    fn ioctl(&self, cmd: IoctlCmd, arg: usize) -> Result<i32> {
-        match cmd {
-            IoctlCmd::TCGETS => {
-                // FIXME: only a work around
-                let tty = get_n_tty();
-                tty.ioctl(cmd, arg)
-            }
-            _ => panic!("Ioctl unsupported"),
-        }
-    }
-
-    fn poll(&self, _mask: IoEvents, _poller: Option<&Poller>) -> IoEvents {
-        IoEvents::empty()
-    }
-
-    fn flush(&self) -> Result<()> {
-        Ok(())
-    }
-
-    fn metadata(&self) -> Metadata {
-        panic!("metadata unsupported");
-    }
-
-    fn seek(&self, seek_from: SeekFrom) -> Result<usize> {
-        panic!("seek unsupported");
-    }
-}

services/libs/jinux-std/src/fs/file_handle/mod.rs

@@ -1,100 +0,0 @@
-//! Opend File Handle
-
-mod file;
-mod inode_handle;
-
-use crate::fs::utils::{IoEvents, Metadata, Poller, SeekFrom};
-use crate::prelude::*;
-use crate::rights::{ReadOp, WriteOp};
-use alloc::sync::Arc;
-
-pub use self::file::File;
-pub use self::inode_handle::InodeHandle;
-
-#[derive(Clone)]
-pub struct FileHandle {
-    inner: Inner,
-}
-
-#[derive(Clone)]
-enum Inner {
-    File(Arc<dyn File>),
-    Inode(InodeHandle),
-}
-
-impl FileHandle {
-    pub fn new_file(file: Arc<dyn File>) -> Arc<Self> {
-        let inner = Inner::File(file);
-        Arc::new(Self { inner })
-    }
-
-    pub fn new_inode_handle(inode_handle: InodeHandle) -> Arc<Self> {
-        let inner = Inner::Inode(inode_handle);
-        Arc::new(Self { inner })
-    }
-
-    pub fn as_file(&self) -> Option<&Arc<dyn File>> {
-        match &self.inner {
-            Inner::File(file) => Some(file),
-            _ => None,
-        }
-    }
-
-    pub fn as_inode_handle(&self) -> Option<&InodeHandle> {
-        match &self.inner {
-            Inner::Inode(inode_handle) => Some(inode_handle),
-            _ => None,
-        }
-    }
-
-    pub fn read(&self, buf: &mut [u8]) -> Result<usize> {
-        match &self.inner {
-            Inner::File(file) => file.read(buf),
-            Inner::Inode(inode_handle) => {
-                let static_handle = inode_handle.clone().to_static::<ReadOp>()?;
-                static_handle.read(buf)
-            }
-        }
-    }
-
-    pub fn write(&self, buf: &[u8]) -> Result<usize> {
-        match &self.inner {
-            Inner::File(file) => file.write(buf),
-            Inner::Inode(inode_handle) => {
-                let static_handle = inode_handle.clone().to_static::<WriteOp>()?;
-                static_handle.write(buf)
-            }
-        }
-    }
-
-    pub fn metadata(&self) -> Metadata {
-        match &self.inner {
-            Inner::File(file) => file.metadata(),
-            Inner::Inode(inode_handle) => inode_handle.dentry().vnode().metadata(),
-        }
-    }
-
-    pub fn seek(&self, seek_from: SeekFrom) -> Result<usize> {
-        match &self.inner {
-            Inner::File(file) => file.seek(seek_from),
-            Inner::Inode(inode_handle) => inode_handle.seek(seek_from),
-        }
-    }
-
-    pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
-        match &self.inner {
-            Inner::File(file) => file.poll(mask, poller),
-            Inner::Inode(inode_handle) => inode_handle.dentry().vnode().poll(mask, poller),
-        }
-    }
-
-    pub fn clean_for_close(&self) -> Result<()> {
-        match &self.inner {
-            Inner::Inode(_) => {
-                // Close does not guarantee that the data has been successfully saved to disk.
-            }
-            Inner::File(file) => file.flush()?,
-        }
-        Ok(())
-    }
-}

services/libs/jinux-std/src/fs/file_table.rs

@@ -5,7 +5,7 @@ use core::cell::Cell;
 use jinux_util::slot_vec::SlotVec;
 
 use super::{
-    file_handle::FileHandle,
+    file_handle::FileLike,
     stdio::{Stderr, Stdin, Stdout},
 };
 
@@ -29,18 +29,9 @@ impl FileTable {
         let stdin = Stdin::new_with_default_console();
         let stdout = Stdout::new_with_default_console();
         let stderr = Stderr::new_with_default_console();
-        table.put(FileTableEntry::new(
+        table.put(FileTableEntry::new(Arc::new(stdin), false));
-            FileHandle::new_file(Arc::new(stdin)),
+        table.put(FileTableEntry::new(Arc::new(stdout), false));
-            false,
+        table.put(FileTableEntry::new(Arc::new(stderr), false));
-        ));
-        table.put(FileTableEntry::new(
-            FileHandle::new_file(Arc::new(stdout)),
-            false,
-        ));
-        table.put(FileTableEntry::new(
-            FileHandle::new_file(Arc::new(stderr)),
-            false,
-        ));
         Self {
             table,
             subject: Subject::new(),
@@ -73,7 +64,7 @@ impl FileTable {
         Ok(min_free_fd as FileDescripter)
     }
 
-    pub fn insert(&mut self, item: Arc<FileHandle>) -> FileDescripter {
+    pub fn insert(&mut self, item: Arc<dyn FileLike>) -> FileDescripter {
         let entry = FileTableEntry::new(item, false);
         self.table.put(entry) as FileDescripter
     }
@@ -81,32 +72,42 @@ impl FileTable {
     pub fn insert_at(
         &mut self,
         fd: FileDescripter,
-        item: Arc<FileHandle>,
+        item: Arc<dyn FileLike>,
-    ) -> Option<Arc<FileHandle>> {
+    ) -> Option<Arc<dyn FileLike>> {
         let entry = FileTableEntry::new(item, false);
         let entry = self.table.put_at(fd as usize, entry);
         if entry.is_some() {
-            self.notify_close_fd_event(fd);
+            let events = FdEvents::Close(fd);
+            self.notify_fd_events(&events);
+            entry.as_ref().unwrap().notify_fd_events(&events);
         }
         entry.map(|e| e.file)
     }
 
-    pub fn close_file(&mut self, fd: FileDescripter) -> Option<Arc<FileHandle>> {
+    pub fn close_file(&mut self, fd: FileDescripter) -> Option<Arc<dyn FileLike>> {
         let entry = self.table.remove(fd as usize);
         if entry.is_some() {
-            self.notify_close_fd_event(fd);
+            let events = FdEvents::Close(fd);
+            self.notify_fd_events(&events);
+            entry.as_ref().unwrap().notify_fd_events(&events);
         }
         entry.map(|e| e.file)
     }
 
-    pub fn get_file(&self, fd: FileDescripter) -> Result<&Arc<FileHandle>> {
+    pub fn get_file(&self, fd: FileDescripter) -> Result<&Arc<dyn FileLike>> {
         self.table
             .get(fd as usize)
             .map(|entry| &entry.file)
             .ok_or(Error::with_message(Errno::EBADF, "fd not exits"))
     }
 
-    pub fn fds_and_files(&self) -> impl Iterator<Item = (FileDescripter, &'_ Arc<FileHandle>)> {
+    pub fn get_entry(&self, fd: FileDescripter) -> Result<&FileTableEntry> {
+        self.table
+            .get(fd as usize)
+            .ok_or(Error::with_message(Errno::EBADF, "fd not exits"))
+    }
+
+    pub fn fds_and_files(&self) -> impl Iterator<Item = (FileDescripter, &'_ Arc<dyn FileLike>)> {
         self.table
             .idxes_and_items()
             .map(|(idx, entry)| (idx as FileDescripter, &entry.file))
@@ -116,13 +117,12 @@ impl FileTable {
         self.subject.register_observer(observer);
     }
 
-    pub fn unregister_observer(&self, observer: Weak<dyn Observer<FdEvents>>) {
+    pub fn unregister_observer(&self, observer: &Weak<dyn Observer<FdEvents>>) {
         self.subject.unregister_observer(observer);
     }
 
-    fn notify_close_fd_event(&self, fd: FileDescripter) {
+    fn notify_fd_events(&self, events: &FdEvents) {
-        let events = FdEvents::Close(fd);
+        self.subject.notify_observers(events);
-        self.subject.notify_observers(&events);
     }
 }
 
@@ -150,17 +150,44 @@ pub enum FdEvents {
 
 impl Events for FdEvents {}
 
-#[derive(Clone)]
+pub struct FileTableEntry {
-struct FileTableEntry {
+    file: Arc<dyn FileLike>,
-    file: Arc<FileHandle>,
     close_on_exec: Cell<bool>,
+    subject: Subject<FdEvents>,
 }
 
 impl FileTableEntry {
-    pub fn new(file: Arc<FileHandle>, close_on_exec: bool) -> Self {
+    pub fn new(file: Arc<dyn FileLike>, close_on_exec: bool) -> Self {
         Self {
             file,
             close_on_exec: Cell::new(close_on_exec),
+            subject: Subject::new(),
+        }
+    }
+
+    pub fn file(&self) -> &Arc<dyn FileLike> {
+        &self.file
+    }
+
+    pub fn register_observer(&self, epoll: Weak<dyn Observer<FdEvents>>) {
+        self.subject.register_observer(epoll);
+    }
+
+    pub fn unregister_observer(&self, epoll: &Weak<dyn Observer<FdEvents>>) {
+        self.subject.unregister_observer(epoll);
+    }
+
+    pub fn notify_fd_events(&self, events: &FdEvents) {
+        self.subject.notify_observers(events);
+    }
+}
+
+impl Clone for FileTableEntry {
+    fn clone(&self) -> Self {
+        Self {
+            file: self.file.clone(),
+            close_on_exec: self.close_on_exec.clone(),
+            subject: Subject::new(),
         }
     }
 }

services/libs/jinux-std/src/fs/fs_resolver.rs

@@ -2,8 +2,8 @@ use crate::prelude::*;
 use alloc::str;
 use alloc::string::String;
 
-use super::file_handle::InodeHandle;
 use super::file_table::FileDescripter;
+use super::inode_handle::InodeHandle;
 use super::procfs::ProcFS;
 use super::ramfs::RamFS;
 use super::utils::{
@@ -253,7 +253,7 @@ impl FsResolver {
         let file_table = current.file_table().lock();
         let inode_handle = file_table
             .get_file(fd)?
-            .as_inode_handle()
+            .downcast_ref::<InodeHandle>()
             .ok_or(Error::with_message(Errno::EBADE, "not inode"))?;
         Ok(inode_handle.dentry().clone())
     }

services/libs/jinux-std/src/fs/inode_handle/dyn_cap.rs

@@ -36,13 +36,6 @@ impl InodeHandle<Rights> {
         Ok(InodeHandle(self.0, R1::new()))
     }
 
-    pub fn read(&self, buf: &mut [u8]) -> Result<usize> {
-        if !self.1.contains(Rights::READ) {
-            return_errno_with_message!(Errno::EBADF, "File is not readable");
-        }
-        self.0.read(buf)
-    }
-
     pub fn read_to_end(&self, buf: &mut Vec<u8>) -> Result<usize> {
         if !self.1.contains(Rights::READ) {
             return_errno_with_message!(Errno::EBADF, "File is not readable");
@@ -50,13 +43,6 @@ impl InodeHandle<Rights> {
         self.0.read_to_end(buf)
     }
 
-    pub fn write(&self, buf: &[u8]) -> Result<usize> {
-        if !self.1.contains(Rights::WRITE) {
-            return_errno_with_message!(Errno::EBADF, "File is not writable");
-        }
-        self.0.write(buf)
-    }
-
     pub fn readdir(&self, visitor: &mut dyn DirentVisitor) -> Result<usize> {
         if !self.1.contains(Rights::READ) {
             return_errno_with_message!(Errno::EBADF, "File is not readable");
@@ -70,3 +56,40 @@ impl Clone for InodeHandle<Rights> {
         Self(self.0.clone(), self.1.clone())
     }
 }
+
+impl FileLike for InodeHandle<Rights> {
+    fn read(&self, buf: &mut [u8]) -> Result<usize> {
+        if !self.1.contains(Rights::READ) {
+            return_errno_with_message!(Errno::EBADF, "File is not readable");
+        }
+        self.0.read(buf)
+    }
+
+    fn write(&self, buf: &[u8]) -> Result<usize> {
+        if !self.1.contains(Rights::WRITE) {
+            return_errno_with_message!(Errno::EBADF, "File is not writable");
+        }
+        self.0.write(buf)
+    }
+
+    fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.dentry().vnode().poll(mask, poller)
+    }
+
+    fn metadata(&self) -> Metadata {
+        self.dentry().vnode().metadata()
+    }
+
+    fn seek(&self, seek_from: SeekFrom) -> Result<usize> {
+        self.0.seek(seek_from)
+    }
+
+    fn clean_for_close(&self) -> Result<()> {
+        // Close does not guarantee that the data has been successfully saved to disk.
+        Ok(())
+    }
+
+    fn as_any_ref(&self) -> &dyn Any {
+        self
+    }
+}

services/libs/jinux-std/src/fs/inode_handle/mod.rs

@@ -3,7 +3,10 @@
 mod dyn_cap;
 mod static_cap;
 
-use crate::fs::utils::{AccessMode, Dentry, DirentVisitor, InodeType, SeekFrom, StatusFlags};
+use crate::fs::file_handle::FileLike;
+use crate::fs::utils::{
+    AccessMode, Dentry, DirentVisitor, InodeType, IoEvents, Metadata, Poller, SeekFrom, StatusFlags,
+};
 use crate::prelude::*;
 use crate::rights::Rights;
 

services/libs/jinux-std/src/fs/mod.rs

@@ -1,7 +1,9 @@
+pub mod epoll;
 pub mod file_handle;
 pub mod file_table;
 pub mod fs_resolver;
 pub mod initramfs;
+pub mod inode_handle;
 pub mod pipe;
 pub mod procfs;
 pub mod ramfs;

services/libs/jinux-std/src/fs/pipe.rs

@@ -1,6 +1,7 @@
+use crate::events::Observer;
 use crate::prelude::*;
 
-use super::file_handle::File;
+use super::file_handle::FileLike;
 use super::utils::{Consumer, IoEvents, Poller, Producer};
 
 pub struct PipeReader {
@@ -13,7 +14,7 @@ impl PipeReader {
     }
 }
 
-impl File for PipeReader {
+impl FileLike for PipeReader {
     fn read(&self, buf: &mut [u8]) -> Result<usize> {
         let is_nonblocking = self.consumer.is_nonblocking();
 
@@ -41,6 +42,25 @@ impl File for PipeReader {
     fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
         self.consumer.poll(mask, poller)
     }
+
+    fn register_observer(
+        &self,
+        observer: Arc<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        self.consumer.register_observer(observer, mask)
+    }
+
+    fn unregister_observer(
+        &self,
+        observer: &Arc<dyn Observer<IoEvents>>,
+    ) -> Result<Arc<dyn Observer<IoEvents>>> {
+        self.consumer.unregister_observer(observer)
+    }
+
+    fn as_any_ref(&self) -> &dyn Any {
+        self
+    }
 }
 
 pub struct PipeWriter {
@@ -53,7 +73,7 @@ impl PipeWriter {
     }
 }
 
-impl File for PipeWriter {
+impl FileLike for PipeWriter {
     fn write(&self, buf: &[u8]) -> Result<usize> {
         let is_nonblocking = self.producer.is_nonblocking();
 
@@ -81,6 +101,25 @@ impl File for PipeWriter {
     fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
         self.producer.poll(mask, poller)
     }
+
+    fn register_observer(
+        &self,
+        observer: Arc<dyn Observer<IoEvents>>,
+        mask: IoEvents,
+    ) -> Result<()> {
+        self.producer.register_observer(observer, mask)
+    }
+
+    fn unregister_observer(
+        &self,
+        observer: &Arc<dyn Observer<IoEvents>>,
+    ) -> Result<Arc<dyn Observer<IoEvents>>> {
+        self.producer.unregister_observer(observer)
+    }
+
+    fn as_any_ref(&self) -> &dyn Any {
+        self
+    }
 }
 
 fn should_io_return(res: &Result<usize>, is_nonblocking: bool) -> bool {

services/libs/jinux-std/src/fs/procfs/pid/fd.rs

@@ -1,6 +1,7 @@
 use super::*;
-use crate::fs::file_handle::FileHandle;
+use crate::fs::file_handle::FileLike;
 use crate::fs::file_table::FileDescripter;
+use crate::fs::inode_handle::InodeHandle;
 
 /// Represents the inode at `/proc/[pid]/fd`.
 pub struct FdDirOps(Arc<Process>);
@@ -62,10 +63,10 @@ impl DirOps for FdDirOps {
 }
 
 /// Represents the inode at `/proc/[pid]/fd/N`.
-struct FileSymOps(Arc<FileHandle>);
+struct FileSymOps(Arc<dyn FileLike>);
 
 impl FileSymOps {
-    pub fn new_inode(file: Arc<FileHandle>, parent: Weak<dyn Inode>) -> Arc<dyn Inode> {
+    pub fn new_inode(file: Arc<dyn FileLike>, parent: Weak<dyn Inode>) -> Arc<dyn Inode> {
         ProcSymBuilder::new(Self(file))
             .parent(parent)
             .build()
@@ -75,13 +76,11 @@ impl FileSymOps {
 
 impl SymOps for FileSymOps {
     fn read_link(&self) -> Result<String> {
-        let path = if let Some(inode_handle) = self.0.as_inode_handle() {
+        let path = if let Some(inode_handle) = self.0.downcast_ref::<InodeHandle>() {
             inode_handle.dentry().abs_path()
-        } else if let Some(file) = self.0.as_file() {
-            // TODO: get the real path for stdio
-            String::from("/dev/tty")
         } else {
-            unreachable!()
+            // TODO: get the real path for other FileLike object
+            String::from("/dev/tty")
         };
         Ok(path)
     }

services/libs/jinux-std/src/fs/stdio.rs

@@ -1,7 +1,7 @@
 use crate::prelude::*;
 use crate::tty::{get_n_tty, Tty};
 
-use super::file_handle::File;
+use super::file_handle::FileLike;
 use super::file_table::FileDescripter;
 use super::utils::{InodeMode, InodeType, IoEvents, Metadata, Poller, SeekFrom};
 
@@ -21,7 +21,7 @@ pub struct Stderr {
     console: Option<Arc<Tty>>,
 }
 
-impl File for Stdin {
+impl FileLike for Stdin {
     fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
         if let Some(console) = self.console.as_ref() {
             console.poll(mask, poller)
@@ -69,8 +69,12 @@ impl File for Stdin {
             rdev: 0,
         }
     }
+
+    fn as_any_ref(&self) -> &dyn Any {
+        self
+    }
 }
-impl File for Stdout {
+impl FileLike for Stdout {
     fn ioctl(&self, cmd: super::utils::IoctlCmd, arg: usize) -> Result<i32> {
         if let Some(console) = self.console.as_ref() {
             console.ioctl(cmd, arg)
@@ -110,9 +114,13 @@ impl File for Stdout {
             rdev: 0,
         }
     }
+
+    fn as_any_ref(&self) -> &dyn Any {
+        self
+    }
 }
 
-impl File for Stderr {
+impl FileLike for Stderr {
     fn ioctl(&self, cmd: super::utils::IoctlCmd, arg: usize) -> Result<i32> {
         if let Some(console) = self.console.as_ref() {
             console.ioctl(cmd, arg)
@@ -152,6 +160,10 @@ impl File for Stderr {
             rdev: 0,
         }
     }
+
+    fn as_any_ref(&self) -> &dyn Any {
+        self
+    }
 }
 
 impl Stdin {

services/libs/jinux-std/src/fs/utils/channel.rs

@@ -2,6 +2,7 @@ use core::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use jinux_rights_proc::require;
 use ringbuf::{HeapConsumer as HeapRbConsumer, HeapProducer as HeapRbProducer, HeapRb};
 
+use crate::events::Observer;
 use crate::prelude::*;
 use crate::rights::*;
 
@@ -48,6 +49,59 @@ pub struct Producer<T>(EndPoint<T, WriteOp>);
 
 pub struct Consumer<T>(EndPoint<T, ReadOp>);
 
+macro_rules! impl_common_methods_for_channel {
+    () => {
+        pub fn shutdown(&self) {
+            self.this_end().shutdown()
+        }
+
+        pub fn is_shutdown(&self) -> bool {
+            self.this_end().is_shutdown()
+        }
+
+        pub fn is_peer_shutdown(&self) -> bool {
+            self.peer_end().is_shutdown()
+        }
+
+        pub fn status_flags(&self) -> StatusFlags {
+            self.this_end().status_flags()
+        }
+
+        pub fn set_status_flags(&self, new_flags: StatusFlags) {
+            self.this_end().set_status_flags(new_flags)
+        }
+
+        pub fn is_nonblocking(&self) -> bool {
+            self.this_end()
+                .status_flags()
+                .contains(StatusFlags::O_NONBLOCK)
+        }
+
+        pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+            self.this_end().pollee.poll(mask, poller)
+        }
+
+        pub fn register_observer(
+            &self,
+            observer: Arc<dyn Observer<IoEvents>>,
+            mask: IoEvents,
+        ) -> Result<()> {
+            self.this_end().pollee.register_observer(observer, mask);
+            Ok(())
+        }
+
+        pub fn unregister_observer(
+            &self,
+            observer: &Arc<dyn Observer<IoEvents>>,
+        ) -> Result<Arc<dyn Observer<IoEvents>>> {
+            self.this_end()
+                .pollee
+                .unregister_observer(observer)
+                .ok_or_else(|| Error::with_message(Errno::ENOENT, "the observer is not registered"))
+        }
+    };
+}
+
 impl<T> Producer<T> {
     fn this_end(&self) -> &EndPointInner<HeapRbProducer<T>> {
         &self.0.common.producer
@@ -75,35 +129,7 @@ impl<T> Producer<T> {
         }
     }
 
-    pub fn shutdown(&self) {
+    impl_common_methods_for_channel!();
-        self.this_end().shutdown()
-    }
-
-    pub fn is_shutdown(&self) -> bool {
-        self.this_end().is_shutdown()
-    }
-
-    pub fn is_peer_shutdown(&self) -> bool {
-        self.peer_end().is_shutdown()
-    }
-
-    pub fn status_flags(&self) -> StatusFlags {
-        self.this_end().status_flags()
-    }
-
-    pub fn set_status_flags(&self, new_flags: StatusFlags) {
-        self.this_end().set_status_flags(new_flags)
-    }
-
-    pub fn is_nonblocking(&self) -> bool {
-        self.this_end()
-            .status_flags()
-            .contains(StatusFlags::O_NONBLOCK)
-    }
-
-    pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
-        self.this_end().pollee.poll(mask, poller)
-    }
 }
 
 impl<T: Copy> Producer<T> {
@@ -167,35 +193,7 @@ impl<T> Consumer<T> {
         }
     }
 
-    pub fn shutdown(&self) {
+    impl_common_methods_for_channel!();
-        self.this_end().shutdown()
-    }
-
-    pub fn is_shutdown(&self) -> bool {
-        self.this_end().is_shutdown()
-    }
-
-    pub fn is_peer_shutdown(&self) -> bool {
-        self.peer_end().is_shutdown()
-    }
-
-    pub fn status_flags(&self) -> StatusFlags {
-        self.this_end().status_flags()
-    }
-
-    pub fn set_status_flags(&self, new_flags: StatusFlags) {
-        self.this_end().set_status_flags(new_flags)
-    }
-
-    pub fn is_nonblocking(&self) -> bool {
-        self.this_end()
-            .status_flags()
-            .contains(StatusFlags::O_NONBLOCK)
-    }
-
-    pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
-        self.this_end().pollee.poll(mask, poller)
-    }
 }
 
 impl<T: Copy> Consumer<T> {

services/libs/jinux-std/src/fs/utils/poll.rs

@@ -73,6 +73,56 @@ impl Pollee {
         }
     }
 
+    /// Register an IoEvents observer.
+    ///
+    /// A registered observer will get notified (through its `on_events` method)
+    /// every time new events specified by the `mask` argument happen on the
+    /// pollee (through the `add_events` method).
+    ///
+    /// If the given observer has already been registered, then its registered
+    /// event mask will be updated.
+    ///
+    /// Note that the observer will always get notified of the events in
+    /// `IoEvents::ALWAYS_POLL` regardless of the value of `mask`.
+    ///
+    /// # Memory leakage
+    ///
+    /// Since an `Arc` for each observer is kept internally by a pollee,
+    /// it is important for the user to call the `unregister_observer` method
+    /// when the observer is no longer interested in the pollee. Otherwise,
+    /// the observer will not be dropped.
+    pub fn register_observer(&self, observer: Arc<dyn Observer<IoEvents>>, mask: IoEvents) {
+        let mut pollers = self.inner.pollers.lock();
+        let is_new = {
+            let observer: KeyableArc<dyn Observer<IoEvents>> = observer.into();
+            let mask = mask | IoEvents::ALWAYS_POLL;
+            pollers.insert(observer, mask).is_none()
+        };
+        if is_new {
+            self.inner.num_pollers.fetch_add(1, Ordering::Release);
+        }
+    }
+
+    /// Unregister an IoEvents observer.
+    ///
+    /// If such an observer is found, then the registered observer will be
+    /// removed from the pollee and returned as the return value. Otherwise,
+    /// a `None` will be returned.
+    pub fn unregister_observer(
+        &self,
+        observer: &Arc<dyn Observer<IoEvents>>,
+    ) -> Option<Arc<dyn Observer<IoEvents>>> {
+        let observer: KeyableArc<dyn Observer<IoEvents>> = observer.clone().into();
+        let mut pollers = self.inner.pollers.lock();
+        let observer = pollers
+            .remove_entry(&observer)
+            .map(|(observer, _)| observer.into());
+        if observer.is_some() {
+            self.inner.num_pollers.fetch_sub(1, Ordering::Relaxed);
+        }
+        observer
+    }
+
     /// Add some events to the pollee's state.
     ///
     /// This method wakes up all registered pollers that are interested in

services/libs/jinux-std/src/prelude.rs

@@ -13,6 +13,7 @@ pub(crate) use alloc::sync::Weak;
 pub(crate) use alloc::vec;
 pub(crate) use alloc::vec::Vec;
 pub(crate) use bitflags::bitflags;
+pub(crate) use core::any::Any;
 pub(crate) use core::ffi::CStr;
 pub(crate) use jinux_frame::config::PAGE_SIZE;
 pub(crate) use jinux_frame::sync::{Mutex, MutexGuard};

services/libs/jinux-std/src/process/process_table.rs

@@ -68,7 +68,7 @@ pub fn register_observer(observer: Weak<dyn Observer<PidEvent>>) {
     PROCESS_TABLE_SUBJECT.register_observer(observer);
 }
 
-pub fn unregister_observer(observer: Weak<dyn Observer<PidEvent>>) {
+pub fn unregister_observer(observer: &Weak<dyn Observer<PidEvent>>) {
     PROCESS_TABLE_SUBJECT.unregister_observer(observer);
 }
 

services/libs/jinux-std/src/syscall/chdir.rs

@@ -1,4 +1,6 @@
-use crate::fs::{file_table::FileDescripter, fs_resolver::FsPath, utils::InodeType};
+use crate::fs::{
+    file_table::FileDescripter, fs_resolver::FsPath, inode_handle::InodeHandle, utils::InodeType,
+};
 use crate::log_syscall_entry;
 use crate::prelude::*;
 use crate::syscall::constants::MAX_FILENAME_LEN;
@@ -38,7 +40,7 @@ pub fn sys_fchdir(fd: FileDescripter) -> Result<SyscallReturn> {
         let file_table = current.file_table().lock();
         let file = file_table.get_file(fd)?;
         let inode_handle = file
-            .as_inode_handle()
+            .downcast_ref::<InodeHandle>()
             .ok_or(Error::with_message(Errno::EBADE, "not inode"))?;
         inode_handle.dentry().clone()
     };

services/libs/jinux-std/src/syscall/epoll.rs

@@ -0,0 +1,133 @@
+use core::time::Duration;
+
+use crate::fs::epoll::{c_epoll_event, EpollCtl, EpollEvent, EpollFile, EpollFlags};
+use crate::fs::file_table::FileDescripter;
+use crate::fs::utils::CreationFlags;
+use crate::log_syscall_entry;
+use crate::prelude::*;
+use crate::util::{read_val_from_user, write_val_to_user};
+
+use super::SyscallReturn;
+use super::{SYS_EPOLL_CREATE1, SYS_EPOLL_CTL, SYS_EPOLL_WAIT};
+
+pub fn sys_epoll_create(size: i32) -> Result<SyscallReturn> {
+    if size <= 0 {
+        return_errno_with_message!(Errno::EINVAL, "size is not positive");
+    }
+    sys_epoll_create1(0)
+}
+
+pub fn sys_epoll_create1(flags: u32) -> Result<SyscallReturn> {
+    log_syscall_entry!(SYS_EPOLL_CREATE1);
+    debug!("flags = 0x{:x}", flags);
+
+    let close_on_exec = {
+        let flags = CreationFlags::from_bits(flags)
+            .ok_or_else(|| Error::with_message(Errno::EINVAL, "invalid flags"))?;
+        if flags == CreationFlags::empty() {
+            false
+        } else if flags == CreationFlags::O_CLOEXEC {
+            true
+        } else {
+            // Only O_CLOEXEC is valid
+            return_errno_with_message!(Errno::EINVAL, "invalid flags");
+        }
+    };
+
+    let current = current!();
+    let epoll_file: Arc<EpollFile> = EpollFile::new();
+    let mut file_table = current.file_table().lock();
+    let fd = file_table.insert(epoll_file);
+    Ok(SyscallReturn::Return(fd as _))
+}
+
+pub fn sys_epoll_ctl(
+    epfd: FileDescripter,
+    op: i32,
+    fd: FileDescripter,
+    event_addr: Vaddr,
+) -> Result<SyscallReturn> {
+    log_syscall_entry!(SYS_EPOLL_CTL);
+    debug!(
+        "epfd = {}, op = {}, fd = {}, event_addr = 0x{:x}",
+        epfd, op, fd, event_addr
+    );
+
+    const EPOLL_CTL_ADD: i32 = 1;
+    const EPOLL_CTL_DEL: i32 = 2;
+    const EPOLL_CTL_MOD: i32 = 3;
+
+    let cmd = match op {
+        EPOLL_CTL_ADD => {
+            let c_epoll_event = read_val_from_user::<c_epoll_event>(event_addr)?;
+            let event = EpollEvent::from(&c_epoll_event);
+            let flags = EpollFlags::from_bits_truncate(c_epoll_event.events);
+            EpollCtl::Add(fd, event, flags)
+        }
+        EPOLL_CTL_DEL => EpollCtl::Del(fd),
+        EPOLL_CTL_MOD => {
+            let c_epoll_event = read_val_from_user::<c_epoll_event>(event_addr)?;
+            let event = EpollEvent::from(&c_epoll_event);
+            let flags = EpollFlags::from_bits_truncate(c_epoll_event.events);
+            EpollCtl::Mod(fd, event, flags)
+        }
+        _ => return_errno_with_message!(Errno::EINVAL, "invalid op"),
+    };
+
+    let current = current!();
+    let file = {
+        let file_table = current.file_table().lock();
+        file_table.get_file(epfd)?.clone()
+    };
+    let epoll_file = file
+        .downcast_ref::<EpollFile>()
+        .ok_or(Error::with_message(Errno::EINVAL, "not epoll file"))?;
+    epoll_file.control(&cmd)?;
+
+    Ok(SyscallReturn::Return(0 as _))
+}
+
+pub fn sys_epoll_wait(
+    epfd: FileDescripter,
+    events_addr: Vaddr,
+    max_events: i32,
+    timeout: i32,
+) -> Result<SyscallReturn> {
+    log_syscall_entry!(SYS_EPOLL_WAIT);
+
+    let max_events = {
+        if max_events <= 0 {
+            return_errno_with_message!(Errno::EINVAL, "max_events is not positive");
+        }
+        max_events as usize
+    };
+    let timeout = if timeout >= 0 {
+        Some(Duration::from_millis(timeout as _))
+    } else {
+        None
+    };
+    debug!(
+        "epfd = {}, events_addr = 0x{:x}, max_events = {}, timeout = {:?}",
+        epfd, events_addr, max_events, timeout
+    );
+
+    let current = current!();
+    let file = {
+        let file_table = current.file_table().lock();
+        file_table.get_file(epfd)?.clone()
+    };
+    let epoll_file = file
+        .downcast_ref::<EpollFile>()
+        .ok_or(Error::with_message(Errno::EINVAL, "not epoll file"))?;
+    let epoll_events = epoll_file.wait(max_events, timeout.as_ref())?;
+
+    // Write back
+    let mut write_addr = events_addr;
+    for epoll_event in epoll_events.iter() {
+        let c_epoll_event = c_epoll_event::from(epoll_event);
+        write_val_to_user(write_addr, &c_epoll_event)?;
+        write_addr += core::mem::size_of::<c_epoll_event>();
+    }
+
+    Ok(SyscallReturn::Return(epoll_events.len() as _))
+}

services/libs/jinux-std/src/syscall/getdents64.rs

@@ -1,5 +1,6 @@
 use crate::fs::{
     file_table::FileDescripter,
+    inode_handle::InodeHandle,
     utils::{DirentVisitor, InodeType},
 };
 use crate::log_syscall_entry;
@@ -27,7 +28,7 @@ pub fn sys_getdents64(
         file_table.get_file(fd)?.clone()
     };
     let inode_handle = file
-        .as_inode_handle()
+        .downcast_ref::<InodeHandle>()
         .ok_or(Error::with_message(Errno::EBADE, "not inode"))?;
     if inode_handle.dentry().inode_type() != InodeType::Dir {
         return_errno!(Errno::ENOTDIR);

services/libs/jinux-std/src/syscall/ioctl.rs

@@ -16,6 +16,6 @@ pub fn sys_ioctl(fd: FileDescripter, cmd: u32, arg: Vaddr) -> Result<SyscallRetu
     let current = current!();
     let file_table = current.file_table().lock();
     let file = file_table.get_file(fd)?;
-    let res = file.as_file().unwrap().ioctl(ioctl_cmd, arg)?;
+    let res = file.ioctl(ioctl_cmd, arg)?;
     return Ok(SyscallReturn::Return(res as _));
 }

services/libs/jinux-std/src/syscall/mod.rs

@@ -9,6 +9,7 @@ use crate::syscall::clock_nanosleep::sys_clock_nanosleep;
 use crate::syscall::clone::sys_clone;
 use crate::syscall::close::sys_close;
 use crate::syscall::dup::{sys_dup, sys_dup2};
+use crate::syscall::epoll::{sys_epoll_create, sys_epoll_create1, sys_epoll_ctl, sys_epoll_wait};
 use crate::syscall::execve::sys_execve;
 use crate::syscall::exit::sys_exit;
 use crate::syscall::exit_group::sys_exit_group;
@@ -76,6 +77,7 @@ mod clone;
 mod close;
 mod constants;
 mod dup;
+mod epoll;
 mod execve;
 mod exit;
 mod exit_group;
@@ -221,10 +223,13 @@ define_syscall_nums!(
     SYS_GETTID = 186,
     SYS_TIME = 201,
     SYS_FUTEX = 202,
+    SYS_EPOLL_CREATE = 213,
     SYS_GETDENTS64 = 217,
     SYS_SET_TID_ADDRESS = 218,
     SYS_CLOCK_NANOSLEEP = 230,
     SYS_EXIT_GROUP = 231,
+    SYS_EPOLL_WAIT = 232,
+    SYS_EPOLL_CTL = 233,
     SYS_TGKILL = 234,
     SYS_WAITID = 247,
     SYS_OPENAT = 257,
@@ -237,6 +242,7 @@ define_syscall_nums!(
     SYS_READLINKAT = 267,
     SYS_SET_ROBUST_LIST = 273,
     SYS_UTIMENSAT = 280,
+    SYS_EPOLL_CREATE1 = 291,
     SYS_PIPE2 = 293,
     SYS_PRLIMIT64 = 302
 );
@@ -355,10 +361,13 @@ pub fn syscall_dispatch(
         SYS_GETTID => syscall_handler!(0, sys_gettid),
         SYS_TIME => syscall_handler!(1, sys_time, args),
         SYS_FUTEX => syscall_handler!(6, sys_futex, args),
+        SYS_EPOLL_CREATE => syscall_handler!(1, sys_epoll_create, args),
         SYS_GETDENTS64 => syscall_handler!(3, sys_getdents64, args),
         SYS_SET_TID_ADDRESS => syscall_handler!(1, sys_set_tid_address, args),
         SYS_CLOCK_NANOSLEEP => syscall_handler!(4, sys_clock_nanosleep, args),
         SYS_EXIT_GROUP => syscall_handler!(1, sys_exit_group, args),
+        SYS_EPOLL_WAIT => syscall_handler!(4, sys_epoll_wait, args),
+        SYS_EPOLL_CTL => syscall_handler!(4, sys_epoll_ctl, args),
         SYS_TGKILL => syscall_handler!(3, sys_tgkill, args),
         SYS_WAITID => syscall_handler!(5, sys_waitid, args),
         SYS_OPENAT => syscall_handler!(4, sys_openat, args),
@@ -371,6 +380,7 @@ pub fn syscall_dispatch(
         SYS_READLINKAT => syscall_handler!(4, sys_readlinkat, args),
         SYS_SET_ROBUST_LIST => syscall_handler!(2, sys_set_robust_list, args),
         SYS_UTIMENSAT => syscall_handler!(4, sys_utimensat, args),
+        SYS_EPOLL_CREATE1 => syscall_handler!(1, sys_epoll_create1, args),
         SYS_PIPE2 => syscall_handler!(2, sys_pipe2, args),
         SYS_PRLIMIT64 => syscall_handler!(4, sys_prlimit64, args),
         _ => {

services/libs/jinux-std/src/syscall/open.rs

@@ -1,5 +1,5 @@
 use crate::fs::{
-    file_handle::{File, FileHandle},
+    file_handle::FileLike,
     file_table::FileDescripter,
     fs_resolver::{FsPath, AT_FDCWD},
 };
@@ -39,7 +39,7 @@ pub fn sys_openat(
 
     if dirfd == AT_FDCWD && pathname == CString::new("./trace")? {
         // Debug use: This file is used for output busybox log
-        let trace_file = FileHandle::new_file(Arc::new(BusyBoxTraceFile) as Arc<dyn File>);
+        let trace_file = Arc::new(BusyBoxTraceFile);
         let current = current!();
         let mut file_table = current.file_table().lock();
         let fd = file_table.insert(trace_file);
@@ -47,7 +47,7 @@ pub fn sys_openat(
     }
 
     if dirfd == AT_FDCWD && pathname == CString::new("/dev/tty")? {
-        let tty_file = FileHandle::new_file(get_n_tty().clone() as Arc<dyn File>);
+        let tty_file = get_n_tty().clone();
         let current = current!();
         let mut file_table = current.file_table().lock();
         let fd = file_table.insert(tty_file);
@@ -60,7 +60,7 @@ pub fn sys_openat(
         let pathname = pathname.to_string_lossy();
         let fs_path = FsPath::new(dirfd, pathname.as_ref())?;
         let inode_handle = current.fs().read().open(&fs_path, flags, mode)?;
-        FileHandle::new_inode_handle(inode_handle)
+        Arc::new(inode_handle)
     };
     let mut file_table = current.file_table().lock();
     let fd = file_table.insert(file_handle);
@@ -74,9 +74,13 @@ pub fn sys_open(pathname_addr: Vaddr, flags: u32, mode: u16) -> Result<SyscallRe
 /// File for output busybox ash log.
 struct BusyBoxTraceFile;
 
-impl File for BusyBoxTraceFile {
+impl FileLike for BusyBoxTraceFile {
     fn write(&self, buf: &[u8]) -> Result<usize> {
         debug!("ASH TRACE: {}", core::str::from_utf8(buf)?);
         Ok(buf.len())
     }
+
+    fn as_any_ref(&self) -> &dyn Any {
+        self
+    }
 }

services/libs/jinux-std/src/syscall/pipe.rs

@@ -1,4 +1,3 @@
-use crate::fs::file_handle::FileHandle;
 use crate::fs::file_table::FileDescripter;
 use crate::fs::pipe::{PipeReader, PipeWriter};
 use crate::fs::utils::{Channel, StatusFlags};
@@ -22,8 +21,8 @@ pub fn sys_pipe2(fds: Vaddr, flags: u32) -> Result<SyscallReturn> {
         .split();
         (PipeReader::new(consumer), PipeWriter::new(producer))
     };
-    let pipe_reader = FileHandle::new_file(Arc::new(reader));
+    let pipe_reader = Arc::new(reader);
-    let pipe_writer = FileHandle::new_file(Arc::new(writer));
+    let pipe_writer = Arc::new(writer);
 
     let current = current!();
     let mut file_table = current.file_table().lock();

services/libs/jinux-std/src/tty/mod.rs

@@ -2,7 +2,7 @@ use self::line_discipline::LineDiscipline;
 use crate::driver::tty::TtyDriver;
 use crate::fs::utils::{InodeMode, InodeType, IoEvents, Metadata};
 use crate::fs::{
-    file_handle::File,
+    file_handle::FileLike,
     utils::{IoctlCmd, Poller},
 };
 use crate::prelude::*;
@@ -51,7 +51,7 @@ impl Tty {
     }
 }
 
-impl File for Tty {
+impl FileLike for Tty {
     fn read(&self, buf: &mut [u8]) -> Result<usize> {
         self.ldisc.read(buf)
     }
@@ -129,6 +129,10 @@ impl File for Tty {
             rdev: 0,
         }
     }
+
+    fn as_any_ref(&self) -> &dyn Any {
+        self
+    }
 }
 
 /// FIXME: should we maintain a static console?