Use one lock for LineDiscipline

kernel/src/device/pty/pty.rs

@@ -27,6 +27,7 @@ use crate::{
 };
 
 const BUFFER_CAPACITY: usize = 4096;
+const IO_CAPACITY: usize = 4096;
 
 /// Pseudo terminal master.
 /// Internally, it has two buffers.
@@ -37,7 +38,6 @@ pub struct PtyMaster {
     index: u32,
     slave: Arc<PtySlave>,
     input: SpinLock<RingBuffer<u8>>,
-    /// The state of input buffer
     pollee: Pollee,
 }
 
@@ -45,8 +45,9 @@ impl PtyMaster {
     pub fn new(ptmx: Arc<dyn Inode>, index: u32) -> Arc<Self> {
         Arc::new_cyclic(move |master| {
             let slave = Arc::new_cyclic(move |weak_self| PtySlave {
-                ldisc: LineDiscipline::new(),
+                ldisc: SpinLock::new(LineDiscipline::new()),
                 job_control: JobControl::new(),
+                pollee: Pollee::new(),
                 master: master.clone(),
                 weak_self: weak_self.clone(),
             });
@@ -73,51 +74,57 @@ impl PtyMaster {
         &self.slave
     }
 
-    pub(super) fn slave_push_char(&self, ch: u8) {
+    fn slave_push(&self, chs: &[u8]) {
-        let mut input = self.input.disable_irq().lock();
+        let mut input = self.input.lock();
-        input.push_overwrite(ch);
+
+        for ch in chs {
+            // TODO: This is termios-specific behavior and should be part of the TTY implementation
+            // instead of the TTY driver implementation. See the ONLCR flag for more details.
+            if *ch == b'\n' {
+                input.push_overwrite(b'\r');
+                input.push_overwrite(b'\n');
+                continue;
+            }
+            input.push_overwrite(*ch);
+        }
         self.pollee.notify(IoEvents::IN);
     }
 
-    pub(super) fn slave_poll(
+    fn slave_echo(&self) -> impl FnMut(&str) + '_ {
-        &self,
+        let mut input = self.input.lock();
-        mask: IoEvents,
+        let mut has_notified = false;
-        mut poller: Option<&mut PollHandle>,
-    ) -> IoEvents {
-        let mut poll_status = IoEvents::empty();
 
-        let poll_in_mask = mask & IoEvents::IN;
+        move |content| {
-        if !poll_in_mask.is_empty() {
+            for byte in content.as_bytes() {
-            let poll_in_status = self.slave.ldisc.poll(poll_in_mask, poller.as_deref_mut());
+                input.push_overwrite(*byte);
-            poll_status |= poll_in_status;
+            }
+
+            if !has_notified {
+                self.pollee.notify(IoEvents::IN);
+                has_notified = true;
+            }
         }
-
-        let poll_out_mask = mask & IoEvents::OUT;
-        if !poll_out_mask.is_empty() {
-            let poll_out_status = self
-                .pollee
-                .poll_with(poll_out_mask, poller, || self.check_io_events());
-            poll_status |= poll_out_status;
-        }
-
-        poll_status
     }
 
-    fn try_read(&self, writer: &mut VmWriter) -> Result<usize> {
+    fn try_read(&self, buf: &mut [u8]) -> Result<usize> {
-        let mut input = self.input.disable_irq().lock();
+        if buf.is_empty() {
+            return Ok(0);
+        }
 
+        let mut input = self.input.lock();
         if input.is_empty() {
             return_errno_with_message!(Errno::EAGAIN, "the buffer is empty");
         }
 
-        let read_len = input.read_fallible(writer)?;
+        let read_len = input.len().min(buf.len());
+        input.pop_slice(&mut buf[..read_len]).unwrap();
         self.pollee.invalidate();
 
         Ok(read_len)
     }
 
     fn check_io_events(&self) -> IoEvents {
-        let input = self.input.disable_irq().lock();
+        let input = self.input.lock();
 
         if !input.is_empty() {
             IoEvents::IN | IoEvents::OUT
@@ -128,60 +135,27 @@ impl PtyMaster {
 }
 
 impl Pollable for PtyMaster {
-    fn poll(&self, mask: IoEvents, mut poller: Option<&mut PollHandle>) -> IoEvents {
+    fn poll(&self, mask: IoEvents, poller: Option<&mut PollHandle>) -> IoEvents {
-        let mut poll_status = IoEvents::empty();
+        self.pollee
-
+            .poll_with(mask, poller, || self.check_io_events())
-        let poll_in_mask = mask & IoEvents::IN;
-        if !poll_in_mask.is_empty() {
-            let poll_in_status = self
-                .pollee
-                .poll_with(poll_in_mask, poller.as_deref_mut(), || {
-                    self.check_io_events()
-                });
-            poll_status |= poll_in_status;
-        }
-
-        let poll_out_mask = mask & IoEvents::OUT;
-        if !poll_out_mask.is_empty() {
-            let poll_out_status = self.slave.ldisc.poll(poll_out_mask, poller);
-            poll_status |= poll_out_status;
-        }
-
-        poll_status
     }
 }
 
 impl FileIo for PtyMaster {
     fn read(&self, writer: &mut VmWriter) -> Result<usize> {
-        if !writer.has_avail() {
+        // TODO: Add support for non-blocking mode and timeout
-            return Ok(0);
+        let mut buf = vec![0u8; writer.avail().min(IO_CAPACITY)];
-        }
+        let read_len = self.wait_events(IoEvents::IN, None, || self.try_read(&mut buf))?;
 
-        // TODO: deal with nonblocking and timeout
+        writer.write_fallible(&mut buf[..read_len].into())?;
-        self.wait_events(IoEvents::IN, None, || self.try_read(writer))
+        Ok(read_len)
     }
 
     fn write(&self, reader: &mut VmReader) -> Result<usize> {
-        let buf = reader.collect()?;
+        let mut buf = vec![0u8; reader.remain().min(IO_CAPACITY)];
-        let write_len = buf.len();
+        let write_len = reader.read_fallible(&mut buf.as_mut_slice().into())?;
-        let mut input = self.input.lock();
-        for character in buf {
-            self.slave.ldisc.push_char(
-                character,
-                |signum| {
-                    if let Some(foreground) = self.slave.job_control.foreground() {
-                        broadcast_signal_async(Arc::downgrade(&foreground), signum);
-                    }
-                },
-                |content| {
-                    for byte in content.as_bytes() {
-                        input.push_overwrite(*byte);
-                    }
-                },
-            );
-        }
 
-        self.pollee.notify(IoEvents::IN);
+        self.slave.master_push(&buf[..write_len]);
         Ok(write_len)
     }
 
@@ -248,8 +222,9 @@ impl Drop for PtyMaster {
 }
 
 pub struct PtySlave {
-    ldisc: LineDiscipline,
+    ldisc: SpinLock<LineDiscipline>,
     job_control: JobControl,
+    pollee: Pollee,
     master: Weak<PtyMaster>,
     weak_self: Weak<Self>,
 }
@@ -262,6 +237,34 @@ impl PtySlave {
     fn master(&self) -> Arc<PtyMaster> {
         self.master.upgrade().unwrap()
     }
+
+    fn master_push(&self, chs: &[u8]) {
+        let mut ldisc = self.ldisc.lock();
+
+        let master = self.master();
+        let mut echo = master.slave_echo();
+
+        for ch in chs {
+            ldisc.push_char(
+                *ch,
+                |signum| {
+                    if let Some(foreground) = self.job_control.foreground() {
+                        broadcast_signal_async(Arc::downgrade(&foreground), signum);
+                    }
+                },
+                &mut echo,
+            );
+        }
+        self.pollee.notify(IoEvents::IN);
+    }
+
+    fn check_io_events(&self) -> IoEvents {
+        if self.ldisc.lock().buffer_len() != 0 {
+            IoEvents::IN | IoEvents::OUT
+        } else {
+            IoEvents::OUT
+        }
+    }
 }
 
 impl Device for PtySlave {
@@ -282,59 +285,57 @@ impl Terminal for PtySlave {
 
 impl Pollable for PtySlave {
     fn poll(&self, mask: IoEvents, poller: Option<&mut PollHandle>) -> IoEvents {
-        self.master().slave_poll(mask, poller)
+        self.pollee
+            .poll_with(mask, poller, || self.check_io_events())
     }
 }
 
 impl FileIo for PtySlave {
     fn read(&self, writer: &mut VmWriter) -> Result<usize> {
-        let mut buf = vec![0u8; writer.avail()];
         self.job_control.wait_until_in_foreground()?;
-        let read_len = self.ldisc.read(&mut buf)?;
+
-        writer.write_fallible(&mut buf.as_slice().into())?;
+        // TODO: Add support for non-blocking mode and timeout
+        let mut buf = vec![0u8; writer.avail().min(IO_CAPACITY)];
+        let read_len =
+            self.wait_events(IoEvents::IN, None, || self.ldisc.lock().try_read(&mut buf))?;
+        self.pollee.invalidate();
+
+        writer.write_fallible(&mut buf[..read_len].into())?;
         Ok(read_len)
     }
 
     fn write(&self, reader: &mut VmReader) -> Result<usize> {
-        let buf = reader.collect()?;
+        let mut buf = vec![0u8; reader.remain().min(IO_CAPACITY)];
-        let write_len = buf.len();
+        let write_len = reader.read_fallible(&mut buf.as_mut_slice().into())?;
-        let master = self.master();
+
-        for ch in buf {
+        self.master().slave_push(&buf[..write_len]);
-            // do we need to add '\r' here?
-            if ch == b'\n' {
-                master.slave_push_char(b'\r');
-                master.slave_push_char(b'\n');
-            } else {
-                master.slave_push_char(ch);
-            }
-        }
         Ok(write_len)
     }
 
     fn ioctl(&self, cmd: IoctlCmd, arg: usize) -> Result<i32> {
         match cmd {
             IoctlCmd::TCGETS => {
-                let termios = self.ldisc.termios();
+                let termios = *self.ldisc.lock().termios();
                 current_userspace!().write_val(arg, &termios)?;
             }
             IoctlCmd::TCSETS => {
                 let termios = current_userspace!().read_val(arg)?;
-                self.ldisc.set_termios(termios);
+                self.ldisc.lock().set_termios(termios);
             }
             IoctlCmd::TIOCGPTN => {
                 let idx = self.index();
                 current_userspace!().write_val(arg, &idx)?;
             }
             IoctlCmd::TIOCGWINSZ => {
-                let winsize = self.ldisc.window_size();
+                let winsize = self.ldisc.lock().window_size();
                 current_userspace!().write_val(arg, &winsize)?;
             }
             IoctlCmd::TIOCSWINSZ => {
                 let winsize = current_userspace!().read_val(arg)?;
-                self.ldisc.set_window_size(winsize);
+                self.ldisc.lock().set_window_size(winsize);
             }
             IoctlCmd::FIONREAD => {
-                let buffer_len = self.ldisc.buffer_len() as i32;
+                let buffer_len = self.ldisc.lock().buffer_len() as i32;
                 current_userspace!().write_val(arg, &buffer_len)?;
             }
             _ => (self.weak_self.upgrade().unwrap() as Arc<dyn Terminal>)

kernel/src/device/tty/line_discipline.rs

@@ -2,16 +2,12 @@
 
 use alloc::format;
 
-use ostd::{sync::LocalIrqDisabled, trap::disable_local};
-
 use super::termio::{KernelTermios, WinSize, CC_C_CHAR};
 use crate::{
-    events::IoEvents,
     prelude::*,
     process::signal::{
         constants::{SIGINT, SIGQUIT},
         sig_num::SigNum,
-        PollHandle, Pollable, Pollee,
     },
     util::ring_buffer::RingBuffer,
 };
@@ -21,22 +17,15 @@ use crate::{
 
 const BUFFER_CAPACITY: usize = 4096;
 
-// Lock ordering to prevent deadlock (circular dependencies):
-// 1. `termios`
-// 2. `current_line`
-// 3. `read_buffer`
-// 4. `work_item_para`
 pub struct LineDiscipline {
     /// Current line
-    current_line: SpinLock<CurrentLine, LocalIrqDisabled>,
+    current_line: CurrentLine,
     /// The read buffer
-    read_buffer: SpinLock<RingBuffer<u8>, LocalIrqDisabled>,
+    read_buffer: RingBuffer<u8>,
     /// Termios
-    termios: SpinLock<KernelTermios, LocalIrqDisabled>,
+    termios: KernelTermios,
     /// Windows size
-    winsize: SpinLock<WinSize, LocalIrqDisabled>,
+    winsize: WinSize,
-    /// Pollee
-    pollee: Pollee,
 }
 
 pub struct CurrentLine {
@@ -78,112 +67,85 @@ impl CurrentLine {
     }
 }
 
-impl Pollable for LineDiscipline {
-    fn poll(&self, mask: IoEvents, poller: Option<&mut PollHandle>) -> IoEvents {
-        self.pollee
-            .poll_with(mask, poller, || self.check_io_events())
-    }
-}
-
 impl LineDiscipline {
-    /// Creates a new line discipline
+    /// Creates a new line discipline.
     pub fn new() -> Self {
         Self {
-            current_line: SpinLock::new(CurrentLine::default()),
+            current_line: CurrentLine::default(),
-            read_buffer: SpinLock::new(RingBuffer::new(BUFFER_CAPACITY)),
+            read_buffer: RingBuffer::new(BUFFER_CAPACITY),
-            termios: SpinLock::new(KernelTermios::default()),
+            termios: KernelTermios::default(),
-            winsize: SpinLock::new(WinSize::default()),
+            winsize: WinSize::default(),
-            pollee: Pollee::new(),
         }
     }
 
-    /// Pushes a char to the line discipline
+    /// Pushes a character to the line discipline.
     pub fn push_char<F1: FnMut(SigNum), F2: FnMut(&str)>(
-        &self,
+        &mut self,
         ch: u8,
         mut signal_callback: F1,
         echo_callback: F2,
     ) {
-        let termios = self.termios.lock();
+        let ch = if self.termios.contains_icrnl() && ch == b'\r' {
-
-        let ch = if termios.contains_icrnl() && ch == b'\r' {
             b'\n'
         } else {
             ch
         };
 
-        if let Some(signum) = char_to_signal(ch, &termios) {
+        if let Some(signum) = char_to_signal(ch, &self.termios) {
             signal_callback(signum);
-            // CBREAK mode may require the character to be outputted, so just go ahead.
+            // CBREAK mode may require the character to be echoed, so just go ahead.
         }
 
-        // Typically, a tty in raw mode does not echo. But the tty can also be in a CBREAK mode,
+        // Typically, a TTY in raw mode does not echo. But the TTY can also be in a CBREAK mode,
         // with ICANON closed and ECHO opened.
-        if termios.contain_echo() {
+        if self.termios.contain_echo() {
-            self.output_char(ch, &termios, echo_callback);
+            self.output_char(ch, echo_callback);
         }
 
         // Raw mode
-        if !termios.is_canonical_mode() {
+        if !self.termios.is_canonical_mode() {
-            self.read_buffer.lock().push_overwrite(ch);
+            self.read_buffer.push_overwrite(ch);
-            self.pollee.notify(IoEvents::IN);
             return;
         }
 
         // Canonical mode
 
-        if ch == *termios.get_special_char(CC_C_CHAR::VKILL) {
+        if ch == *self.termios.get_special_char(CC_C_CHAR::VKILL) {
             // Erase current line
-            self.current_line.lock().drain();
+            self.current_line.drain();
         }
 
-        if ch == *termios.get_special_char(CC_C_CHAR::VERASE) {
+        if ch == *self.termios.get_special_char(CC_C_CHAR::VERASE) {
             // Type backspace
-            let mut current_line = self.current_line.lock();
+            self.current_line.backspace();
-            if !current_line.is_empty() {
-                current_line.backspace();
-            }
         }
 
-        if is_line_terminator(ch, &termios) {
+        if is_line_terminator(ch, &self.termios) {
-            // If a new line is met, all bytes in current_line will be moved to read_buffer
+            // A new line is met. Move all bytes in `current_line` to `read_buffer`.
-            let mut current_line = self.current_line.lock();
+            self.current_line.push_char(ch);
-            current_line.push_char(ch);
+            for line_ch in self.current_line.drain() {
-            let current_line_chars = current_line.drain();
+                self.read_buffer.push_overwrite(line_ch);
-            for char in current_line_chars {
-                self.read_buffer.lock().push_overwrite(char);
-                self.pollee.notify(IoEvents::IN);
             }
         }
 
         if is_printable_char(ch) {
             // Printable character
-            self.current_line.lock().push_char(ch);
+            self.current_line.push_char(ch);
-        }
-    }
-
-    fn check_io_events(&self) -> IoEvents {
-        let buffer = self.read_buffer.lock();
-
-        if !buffer.is_empty() {
-            IoEvents::IN
-        } else {
-            IoEvents::empty()
         }
     }
 
     // TODO: respect output flags
-    fn output_char<F: FnMut(&str)>(&self, ch: u8, termios: &KernelTermios, mut echo_callback: F) {
+    fn output_char<F: FnMut(&str)>(&self, ch: u8, mut echo_callback: F) {
         match ch {
             b'\n' => echo_callback("\n"),
             b'\r' => echo_callback("\r\n"),
-            ch if ch == *termios.get_special_char(CC_C_CHAR::VERASE) => {
+            ch if ch == *self.termios.get_special_char(CC_C_CHAR::VERASE) => {
-                // write a space to overwrite current character
+                // Write a space to overwrite the current character
                 let backspace: &str = core::str::from_utf8(b"\x08 \x08").unwrap();
                 echo_callback(backspace);
             }
             ch if is_printable_char(ch) => print!("{}", char::from(ch)),
-            ch if is_ctrl_char(ch) && termios.contains_echo_ctl() => {
+            ch if is_ctrl_char(ch) && self.termios.contains_echo_ctl() => {
                 let ctrl_char = format!("^{}", ctrl_char_to_printable(ch));
                 echo_callback(&ctrl_char);
             }
@@ -191,140 +153,86 @@ impl LineDiscipline {
         }
     }
 
-    pub fn read(&self, buf: &mut [u8]) -> Result<usize> {
-        self.wait_events(IoEvents::IN, None, || self.try_read(buf))
-    }
-
-    /// Reads all bytes buffered to `dst`.
-    ///
-    /// This method returns the actual read length.
-    fn try_read(&self, dst: &mut [u8]) -> Result<usize> {
-        let (vmin, vtime) = {
-            let termios = self.termios.lock();
-            let vmin = *termios.get_special_char(CC_C_CHAR::VMIN);
-            let vtime = *termios.get_special_char(CC_C_CHAR::VTIME);
-            (vmin, vtime)
-        };
-        let read_len = {
-            if vmin == 0 && vtime == 0 {
-                // poll read
-                self.poll_read(dst)
-            } else if vmin > 0 && vtime == 0 {
-                // block read
-                self.block_read(dst, vmin)?
-            } else if vmin == 0 && vtime > 0 {
-                todo!()
-            } else if vmin > 0 && vtime > 0 {
-                todo!()
-            } else {
-                unreachable!()
-            }
-        };
-        self.pollee.invalidate();
-        Ok(read_len)
-    }
-
     /// Reads bytes from `self` to `dst`, returning the actual bytes read.
     ///
-    /// If no bytes are available, this method returns 0 immediately.
-    fn poll_read(&self, dst: &mut [u8]) -> usize {
-        let termios = self.termios.lock();
-        let mut buffer = self.read_buffer.lock();
-        let len = buffer.len();
-        let max_read_len = len.min(dst.len());
-        if max_read_len == 0 {
-            return 0;
-        }
-        let mut read_len = 0;
-        for dst_i in dst.iter_mut().take(max_read_len) {
-            if let Some(next_char) = buffer.pop() {
-                if termios.is_canonical_mode() {
-                    // canonical mode, read until meet new line
-                    if is_line_terminator(next_char, &termios) {
-                        // The eof should not be read
-                        if !is_eof(next_char, &termios) {
-                            *dst_i = next_char;
-                            read_len += 1;
-                        }
-                        break;
-                    } else {
-                        *dst_i = next_char;
-                        read_len += 1;
-                    }
-                } else {
-                    // raw mode
-                    // FIXME: avoid additional bound check
-                    *dst_i = next_char;
-                    read_len += 1;
-                }
-            } else {
-                break;
-            }
-        }
-
-        read_len
-    }
-
-    /// Reads bytes from `self` into `dst`,
-    /// returning the actual number of bytes read.
-    ///
     /// # Errors
     ///
-    /// If the available bytes are fewer than `min(dst.len(), vmin)`,
+    /// If no bytes are available or the available bytes are fewer than `min(dst.len(), vmin)`,
     /// this method returns [`Errno::EAGAIN`].
-    pub fn block_read(&self, dst: &mut [u8], vmin: u8) -> Result<usize> {
+    pub fn try_read(&mut self, dst: &mut [u8]) -> Result<usize> {
-        let _guard = disable_local();
+        let vmin = *self.termios.get_special_char(CC_C_CHAR::VMIN);
-        let buffer_len = self.read_buffer.lock().len();
+        let vtime = *self.termios.get_special_char(CC_C_CHAR::VTIME);
-        if buffer_len >= dst.len() {
+
-            return Ok(self.poll_read(dst));
+        if vtime != 0 {
+            warn!("non-zero VTIME is not supported");
         }
-        if buffer_len < vmin as usize {
+
-            return_errno!(Errno::EAGAIN);
+        // If `vmin` is zero or `dst` is empty, the following condition will always be false. This
+        // is correct, as the expected behavior is to never block or return `EAGAIN`.
+        if self.buffer_len() < dst.len().min(vmin as _) {
+            return_errno_with_message!(
+                Errno::EAGAIN,
+                "the characters in the buffer are not enough"
+            );
         }
-        Ok(self.poll_read(&mut dst[..buffer_len]))
+
+        for (i, dst_i) in dst.iter_mut().enumerate() {
+            let Some(ch) = self.read_buffer.pop() else {
+                // No more characters
+                return Ok(i);
+            };
+
+            if self.termios.is_canonical_mode() && is_eof(ch, &self.termios) {
+                // This allows the userspace program to see `Ok(0)`
+                return Ok(i);
+            }
+
+            *dst_i = ch;
+
+            if self.termios.is_canonical_mode() && is_line_terminator(ch, &self.termios) {
+                // Read until line terminators in canonical mode
+                return Ok(i + 1);
+            }
+        }
+
+        Ok(dst.len())
     }
 
-    /// Returns whether there is buffered data
+    pub fn termios(&self) -> &KernelTermios {
-    pub fn is_empty(&self) -> bool {
+        &self.termios
-        self.read_buffer.lock().len() == 0
     }
 
-    pub fn termios(&self) -> KernelTermios {
+    pub fn set_termios(&mut self, termios: KernelTermios) {
-        *self.termios.lock()
+        self.termios = termios;
     }
 
-    pub fn set_termios(&self, termios: KernelTermios) {
+    pub fn drain_input(&mut self) {
-        *self.termios.lock() = termios;
+        self.current_line.drain();
-    }
+        self.read_buffer.clear();
-
-    pub fn drain_input(&self) {
-        self.current_line.lock().drain();
-        self.read_buffer.lock().clear();
-        self.pollee.invalidate();
     }
 
     pub fn buffer_len(&self) -> usize {
-        self.read_buffer.lock().len()
+        self.read_buffer.len()
     }
 
     pub fn window_size(&self) -> WinSize {
-        *self.winsize.lock()
+        self.winsize
     }
 
-    pub fn set_window_size(&self, winsize: WinSize) {
+    pub fn set_window_size(&mut self, winsize: WinSize) {
-        *self.winsize.lock() = winsize;
+        self.winsize = winsize;
     }
 }
 
-fn is_line_terminator(item: u8, termios: &KernelTermios) -> bool {
+fn is_line_terminator(ch: u8, termios: &KernelTermios) -> bool {
-    if item == b'\n'
+    if ch == b'\n'
-        || item == *termios.get_special_char(CC_C_CHAR::VEOF)
+        || ch == *termios.get_special_char(CC_C_CHAR::VEOF)
-        || item == *termios.get_special_char(CC_C_CHAR::VEOL)
+        || ch == *termios.get_special_char(CC_C_CHAR::VEOL)
     {
         return true;
     }
 
-    if termios.contains_iexten() && item == *termios.get_special_char(CC_C_CHAR::VEOL2) {
+    if termios.contains_iexten() && ch == *termios.get_special_char(CC_C_CHAR::VEOL2) {
         return true;
     }
 

kernel/src/device/tty/mod.rs

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MPL-2.0
 
-use ostd::early_print;
+use ostd::{early_print, sync::LocalIrqDisabled};
 use spin::Once;
 
 use self::{driver::TtyDriver, line_discipline::LineDiscipline};
@@ -15,7 +15,7 @@ use crate::{
     prelude::*,
     process::{
         broadcast_signal_async,
-        signal::{PollHandle, Pollable},
+        signal::{PollHandle, Pollable, Pollee},
         JobControl, Terminal,
     },
 };
@@ -36,13 +36,16 @@ pub(super) fn init() {
     driver::init();
 }
 
+const IO_CAPACITY: usize = 4096;
+
 pub struct Tty {
     /// tty_name
     #[expect(unused)]
     name: CString,
     /// line discipline
-    ldisc: LineDiscipline,
+    ldisc: SpinLock<LineDiscipline, LocalIrqDisabled>,
     job_control: JobControl,
+    pollee: Pollee,
     /// driver
     driver: SpinLock<Weak<TtyDriver>>,
     weak_self: Weak<Self>,
@@ -52,8 +55,9 @@ impl Tty {
     pub fn new(name: CString) -> Arc<Self> {
         Arc::new_cyclic(move |weak_ref| Tty {
             name,
-            ldisc: LineDiscipline::new(),
+            ldisc: SpinLock::new(LineDiscipline::new()),
             job_control: JobControl::new(),
+            pollee: Pollee::new(),
             driver: SpinLock::new(Weak::new()),
             weak_self: weak_ref.clone(),
         })
@@ -66,7 +70,7 @@ impl Tty {
     pub fn push_char(&self, ch: u8) {
         // FIXME: Use `early_print` to avoid calling virtio-console.
         // This is only a workaround
-        self.ldisc.push_char(
+        self.ldisc.lock().push_char(
             ch,
             |signum| {
                 if let Some(foreground) = self.job_control.foreground() {
@@ -74,69 +78,90 @@ impl Tty {
                 }
             },
             |content| early_print!("{}", content),
-        )
+        );
+        self.pollee.notify(IoEvents::IN);
+    }
+
+    fn check_io_events(&self) -> IoEvents {
+        if self.ldisc.lock().buffer_len() != 0 {
+            IoEvents::IN | IoEvents::OUT
+        } else {
+            IoEvents::OUT
+        }
     }
 }
 
 impl Pollable for Tty {
     fn poll(&self, mask: IoEvents, poller: Option<&mut PollHandle>) -> IoEvents {
-        self.ldisc.poll(mask, poller)
+        self.pollee
+            .poll_with(mask, poller, || self.check_io_events())
     }
 }
 
 impl FileIo for Tty {
     fn read(&self, writer: &mut VmWriter) -> Result<usize> {
-        let mut buf = vec![0; writer.avail()];
         self.job_control.wait_until_in_foreground()?;
-        let read_len = self.ldisc.read(buf.as_mut_slice())?;
+
-        writer.write_fallible(&mut buf.as_slice().into())?;
+        // TODO: Add support for non-blocking mode and timeout
+        let mut buf = vec![0u8; writer.avail().min(IO_CAPACITY)];
+        let read_len =
+            self.wait_events(IoEvents::IN, None, || self.ldisc.lock().try_read(&mut buf))?;
+        self.pollee.invalidate();
+
+        // TODO: Confirm what we should do if `write_fallible` fails in the middle.
+        writer.write_fallible(&mut buf[..read_len].into())?;
         Ok(read_len)
     }
 
     fn write(&self, reader: &mut VmReader) -> Result<usize> {
-        let buf = reader.collect()?;
+        let mut buf = vec![0u8; reader.remain().min(IO_CAPACITY)];
-        if let Ok(content) = alloc::str::from_utf8(&buf) {
+        let write_len = reader.read_fallible(&mut buf.as_mut_slice().into())?;
+
+        if let Ok(content) = alloc::str::from_utf8(&buf[..write_len]) {
             print!("{content}");
         } else {
             println!("Not utf-8 content: {:?}", buf);
         }
-        Ok(buf.len())
+        Ok(write_len)
     }
 
     fn ioctl(&self, cmd: IoctlCmd, arg: usize) -> Result<i32> {
         match cmd {
             IoctlCmd::TCGETS => {
-                // Get terminal attributes
+                let termios = *self.ldisc.lock().termios();
-                let termios = self.ldisc.termios();
+
-                trace!("get termios = {:?}", termios);
                 current_userspace!().write_val(arg, &termios)?;
             }
             IoctlCmd::TCSETS => {
-                // Set terminal attributes
                 let termios = current_userspace!().read_val(arg)?;
-                debug!("set termios = {:?}", termios);
+
-                self.ldisc.set_termios(termios);
+                self.ldisc.lock().set_termios(termios);
             }
             IoctlCmd::TCSETSW => {
                 let termios = current_userspace!().read_val(arg)?;
-                debug!("set termios = {:?}", termios);
+
-                self.ldisc.set_termios(termios);
+                self.ldisc.lock().set_termios(termios);
-                // TODO: drain output buffer
+                // TODO: Drain the output buffer
             }
             IoctlCmd::TCSETSF => {
                 let termios = current_userspace!().read_val(arg)?;
-                debug!("set termios = {:?}", termios);
+
-                self.ldisc.set_termios(termios);
+                let mut ldisc = self.ldisc.lock();
-                self.ldisc.drain_input();
+                ldisc.set_termios(termios);
-                // TODO: drain output buffer
+                ldisc.drain_input();
+                // TODO: Drain the output buffer
+
+                self.pollee.invalidate();
             }
             IoctlCmd::TIOCGWINSZ => {
-                let winsize = self.ldisc.window_size();
+                let winsize = self.ldisc.lock().window_size();
+
                 current_userspace!().write_val(arg, &winsize)?;
             }
             IoctlCmd::TIOCSWINSZ => {
                 let winsize = current_userspace!().read_val(arg)?;
-                self.ldisc.set_window_size(winsize);
+
+                self.ldisc.lock().set_window_size(winsize);
             }
             _ => (self.weak_self.upgrade().unwrap() as Arc<dyn Terminal>)
                 .job_ioctl(cmd, arg, false)?,