Add timeout parameter for poller.wait

services/libs/jinux-std/src/device/pty/pty.rs

@@ -105,7 +105,8 @@ impl FileLike for PtyMaster {
 
                 if events.is_empty() {
                     drop(input);
-                    poller.wait();
+                    // FIXME: deal with pty read timeout
+                    poller.wait(None)?;
                 }
                 continue;
             }

services/libs/jinux-std/src/device/tty/line_discipline.rs

@@ -252,7 +252,8 @@ impl LineDiscipline {
             };
             let revents = self.pollee.poll(IoEvents::IN, need_poller);
             if revents.is_empty() {
-                poller.as_ref().unwrap().wait();
+                // FIXME: deal with ldisc read timeout
+                poller.as_ref().unwrap().wait(None)?;
             }
         }
     }

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

@@ -191,8 +191,7 @@ impl EpollFile {
                 }
             }
 
-            // FIXME: respect timeout parameter
+            poller.as_ref().unwrap().wait(timeout)?;
-            poller.as_ref().unwrap().wait();
         }
     }
 

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

@@ -152,7 +152,8 @@ impl<T: Copy> Producer<T> {
             }
             let events = self.poll(mask, Some(&poller));
             if events.is_empty() {
-                poller.wait();
+                // FIXME: should channel deal with timeout?
+                poller.wait(None)?;
             }
         }
     }
@@ -240,7 +241,8 @@ impl<T: Copy> Consumer<T> {
             }
             let events = self.poll(mask, Some(&poller));
             if events.is_empty() {
-                poller.wait();
+                // FIXME: should channel have timeout?
+                poller.wait(None)?;
             }
         }
     }

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

@@ -3,6 +3,7 @@ use crate::events::{Observer, Subject};
 use crate::prelude::*;
 
 use core::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
+use core::time::Duration;
 use jinux_frame::sync::WaitQueue;
 use keyable_arc::KeyableWeak;
 
@@ -157,8 +158,9 @@ impl Poller {
     }
 
     /// Wait until there are any interesting events happen since last `wait`.
-    pub fn wait(&self) {
+    pub fn wait(&self, timeout: Option<&Duration>) -> Result<()> {
-        self.inner.event_counter.read();
+        self.inner.event_counter.read(timeout)?;
+        Ok(())
     }
 
     fn observer(&self) -> Weak<dyn Observer<IoEvents>> {
@@ -202,20 +204,19 @@ impl EventCounter {
         }
     }
 
-    pub fn read(&self) -> usize {
+    pub fn read(&self, timeout: Option<&Duration>) -> Result<usize> {
-        self.wait_queue
+        let val = self.wait_queue.wait_until(
-            .wait_until(
+            || {
-                || {
+                let val = self.counter.swap(0, Ordering::Relaxed);
-                    let val = self.counter.swap(0, Ordering::Relaxed);
+                if val > 0 {
-                    if val > 0 {
+                    Some(val)
-                        Some(val)
+                } else {
-                    } else {
+                    None
-                        None
+                }
-                    }
+            },
-                },
+            timeout,
-                None,
+        )?;
-            )
+        Ok(val)
-            .unwrap()
     }
 
     pub fn write(&self) {

services/libs/jinux-std/src/net/socket/ip/datagram.rs

@@ -258,7 +258,8 @@ impl Socket for DatagramSocket {
                 if self.nonblocking() {
                     return_errno_with_message!(Errno::EAGAIN, "try to receive again");
                 }
-                poller.wait();
+                // FIXME: deal with recvfrom timeout
+                poller.wait(None)?;
             }
         }
     }

services/libs/jinux-std/src/net/socket/ip/stream/connected.rs

@@ -57,7 +57,8 @@ impl ConnectedStream {
                 if self.is_nonblocking() {
                     return_errno_with_message!(Errno::EAGAIN, "try to recv again");
                 }
-                poller.wait();
+                // FIXME: deal with receive timeout
+                poller.wait(None)?;
             }
         }
     }

services/libs/jinux-std/src/net/socket/ip/stream/init.rs

@@ -151,7 +151,8 @@ impl InitStream {
             } else if self.is_nonblocking() {
                 return_errno_with_message!(Errno::EAGAIN, "try connect again");
             } else {
-                poller.wait();
+                // FIXME: deal with connecting timeout
+                poller.wait(None)?;
             }
         }
     }

services/libs/jinux-std/src/net/socket/ip/stream/listen.rs

@@ -45,7 +45,8 @@ impl ListenStream {
                     if self.is_nonblocking() {
                         return_errno_with_message!(Errno::EAGAIN, "try accept again");
                     }
-                    poller.wait();
+                    // FIXME: deal with accept timeout
+                    poller.wait(None)?;
                 }
                 continue;
             };

services/libs/jinux-std/src/net/socket/unix/stream/listener.rs

@@ -131,8 +131,9 @@ impl BacklogTable {
                 return_errno_with_message!(Errno::ECONNABORTED, "connection is aborted");
             }
 
+            // FIXME: deal with accept timeout
             if events.is_empty() {
-                poller.wait();
+                poller.wait(None)?;
             }
         }
     }

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

@@ -91,8 +91,7 @@ pub fn do_poll(poll_fds: &[PollFd], timeout: Option<Duration>) -> Result<usize>
             return Ok(0);
         }
 
-        // FIXME: respect timeout parameter
+        poller.wait(timeout.as_ref())?;
-        poller.wait();
     }
 }