Support events

examples/chan.rs

@@ -23,17 +23,21 @@ fn main() -> io::Result<()> {
 
     dup.write(b"abc")?;
     dup.flush()?;
+    println!("-> Wrote message");
 
     let mut buf = [0; 5];
     assert_eq!(client.read(&mut buf)?, 3);
     assert_eq!(&buf[..3], b"abc");
+    println!("-> Read message");
 
     println!("Testing blocking I/O...");
 
     let mut client_clone = client.try_clone()?;
 
     let thread = thread::spawn(move || -> io::Result<()> {
+        println!("--> Thread: Sleeping for 1 second...");
         thread::sleep(Duration::from_secs(1));
+        println!("--> Thread: Writing...");
         client_clone.write(b"def")?;
         client_clone.flush()?;
         Ok(())
@@ -41,6 +45,7 @@ fn main() -> io::Result<()> {
 
     assert_eq!(dup.read(&mut buf)?, 3);
     assert_eq!(&buf[..3], b"def");
+    println!("-> Read message");
 
     thread.join().unwrap().unwrap();
 
@@ -50,7 +55,59 @@ fn main() -> io::Result<()> {
         .map_err(from_syscall_error)?;
     assert_eq!(client.read(&mut buf).unwrap_err().kind(), io::ErrorKind::WouldBlock);
 
-    println!("It works!");
+    println!("Testing events...");
+
+    let thread = thread::spawn(move || -> io::Result<()> {
+        println!("--> Thread: Sleeping for 1 second...");
+        thread::sleep(Duration::from_secs(1));
+        println!("--> Thread: Writing...");
+        dup.write(b"hello")?;
+        dup.flush()?;
+        Ok(())
+    });
+
+    let mut event_file = File::open("event:")?;
+    let mut time_file = File::open(format!("time:{}", syscall::CLOCK_MONOTONIC))?;
+
+    let mut time = syscall::TimeSpec::default();
+    time_file.read(&mut time)?;
+    time.tv_sec += 5;
+    time_file.write(&time)?;
+
+    event_file.write(&syscall::Event {
+        id: client.as_raw_fd(),
+        flags: syscall::EVENT_READ | syscall::EVENT_WRITE,
+        data: 0
+    })?;
+    event_file.write(&syscall::Event {
+        id: time_file.as_raw_fd(),
+        flags: syscall::EVENT_READ,
+        data: 1
+    })?;
+
+    let mut event = syscall::Event::default();
+
+    event_file.read(&mut event)?;
+    assert_eq!(event.id, client.as_raw_fd());
+    assert_eq!(event.flags, syscall::EVENT_WRITE);
+    assert_eq!(event.data, 0);
+    println!("-> Read event");
+
+    event_file.read(&mut event)?;
+    assert_eq!(event.id, client.as_raw_fd());
+    assert_eq!(event.flags, syscall::EVENT_READ);
+    assert_eq!(event.data, 0);
+    println!("-> Read event");
+
+    event_file.read(&mut event)?;
+    assert_eq!(event.id, time_file.as_raw_fd());
+    assert_eq!(event.flags, syscall::EVENT_READ);
+    assert_eq!(event.data, 1);
+    println!("-> Timed out");
+
+    thread.join().unwrap().unwrap();
+
+    println!("Everything tested!");
 
     Ok(())
 }

src/main.rs

@@ -50,8 +50,21 @@ fn main() -> Result<(), Box<::std::error::Error>> {
 
             true
         });
+
         if let Some(err) = error {
             return Err(Box::new(err));
         }
+
+        scheme.post_fevents(&mut scheme_file)?;
     }
 }
+fn post_fevent(file: &mut File, id: usize, flag: usize) -> io::Result<()> {
+    file.write(&syscall::Packet {
+        a: syscall::SYS_FEVENT,
+        b: id,
+        c: flag,
+        d: 1,
+        ..Default::default()
+    })
+    .map(|_| ())
+}

src/scheme.rs

-use std::collections::BTreeMap;
+use post_fevent;
+use std::{
+    collections::BTreeMap,
+    fs::File,
+    io
+};
 use syscall::{flag::*, error::*, Error, SchemeBlockMut, Result};
 
+#[derive(Default)]
 pub struct Handle {
     flags: usize,
+    fevent: usize,
+    notified_read: bool,
+    notified_write: bool,
+
     path: Option<String>,
     remote: Option<usize>,
     buffer: Vec<u8>
 }
 impl Handle {
     pub fn accept(&mut self) -> Self {
-        Self {
-            flags: 0,
-            path: None,
-            remote: self.remote.take(),
-            buffer: Vec::new()
-        }
-    }
-    pub fn copy(&self) -> Self {
         Self {
             flags: self.flags,
-            path: None,
-            remote: self.remote,
-            buffer: Vec::new()
+            remote: self.remote.take(),
+            ..Default::default()
         }
     }
 }
@@ -32,6 +33,25 @@ pub struct ChanScheme {
     listeners: BTreeMap<String, usize>,
     next_id: usize
 }
+impl ChanScheme {
+    pub fn post_fevents(&mut self, file: &mut File) -> io::Result<()> {
+        for (id, handle) in &mut self.handles {
+            if !handle.notified_write {
+                handle.notified_write = true;
+                post_fevent(file, *id, EVENT_WRITE)?;
+            }
+            if !handle.buffer.is_empty() {
+                if !handle.notified_read {
+                    handle.notified_read = true;
+                    post_fevent(file, *id, EVENT_READ)?;
+                }
+            } else {
+                handle.notified_read = false;
+            }
+        }
+        Ok(())
+    }
+}
 impl SchemeBlockMut for ChanScheme {
     fn open(&mut self, path: &[u8], flags: usize, _uid: u32, _gid: u32) -> Result<Option<usize>> {
         let path = ::std::str::from_utf8(path).unwrap_or("");
@@ -44,12 +64,8 @@ impl SchemeBlockMut for ChanScheme {
             return Err(Error::new(ENOENT));
         }
 
-        let mut handle = Handle {
-            flags: 0,
-            path: None,
-            remote: None,
-            buffer: Vec::new()
-        };
+        let mut handle = Handle::default();
+        handle.flags = flags;
 
         let id = self.next_id;
         if flags & O_CREAT == O_CREAT {
@@ -67,11 +83,11 @@ impl SchemeBlockMut for ChanScheme {
     fn dup(&mut self, id: usize, buf: &[u8]) -> Result<Option<usize>> {
         match buf {
             b"listen" => {
-                let mut handle = match self.handles.get(&id) {
-                    Some(ref handle) if handle.path.is_some() => handle.copy(),
+                let (flags, remote) = match self.handles.get(&id) {
+                    Some(ref handle) if handle.path.is_some() => (handle.flags, handle.remote),
                     _ => return Err(Error::new(EBADF))
                 };
-                if let Some(remote) = handle.remote {
+                if let Some(remote) = remote {
                     let new_id = self.next_id;
                     let mut clone = self.handles.get_mut(&id).map(Handle::accept).unwrap();
 
@@ -81,7 +97,7 @@ impl SchemeBlockMut for ChanScheme {
                     let mut remote = self.handles.get_mut(&remote).unwrap();
                     remote.remote = Some(new_id);
                     Ok(Some(new_id))
-                } else if handle.flags & O_NONBLOCK == O_NONBLOCK {
+                } else if flags & O_NONBLOCK == O_NONBLOCK {
                     Err(Error::new(EAGAIN))
                 } else {
                     Ok(None)
@@ -95,8 +111,8 @@ impl SchemeBlockMut for ChanScheme {
     fn fcntl(&mut self, id: usize, cmd: usize, arg: usize) -> Result<Option<usize>> {
         match self.handles.get_mut(&id) {
             Some(handle) => match cmd {
-                ::syscall::F_GETFL => Ok(Some(handle.flags)),
-                ::syscall::F_SETFL => {
+                F_GETFL => Ok(Some(handle.flags)),
+                F_SETFL => {
                     handle.flags = arg;
                     Ok(Some(0))
                 },
@@ -105,6 +121,17 @@ impl SchemeBlockMut for ChanScheme {
             _ => Err(Error::new(EBADF))
         }
     }
+    fn fevent(&mut self, id: usize, flags: usize) -> Result<Option<usize>> {
+        match self.handles.get_mut(&id) {
+            Some(handle) => {
+                handle.fevent = flags;
+                handle.notified_read = false;
+                handle.notified_write = false;
+                Ok(Some(id))
+            },
+            _ => Err(Error::new(EBADF))
+        }
+    }
     fn write(&mut self, id: usize, buf: &[u8]) -> Result<Option<usize>> {
         let remote = match self.handles.get(&id) {
             Some(handle) if handle.path.is_none() => handle.remote,