diff --git a/examples/logserver.rs b/examples/logserver.rs index 708e11d363..1a2159a519 100644 --- a/examples/logserver.rs +++ b/examples/logserver.rs @@ -9,17 +9,14 @@ use std::{ fn from_syscall_error(error: syscall::Error) -> io::Error { io::Error::from_raw_os_error(error.errno as i32) } -fn nonblock(file: &File) -> io::Result<()> { - syscall::fcntl(file.as_raw_fd() as usize, syscall::F_SETFL, syscall::O_NONBLOCK) - .map(|_| ()) - .map_err(from_syscall_error) -} fn main() -> io::Result<()> { let server = File::create("chan:log")?; let mut event_file = File::open("event:")?; - nonblock(&server)?; + syscall::fcntl(server.as_raw_fd() as usize, syscall::F_SETFL, syscall::O_NONBLOCK) + .map(|_| ()) + .map_err(from_syscall_error)?; event_file.write(&syscall::Event { id: server.as_raw_fd() as usize, data: 0, @@ -36,41 +33,28 @@ fn main() -> io::Result<()> { if event.data == 0 { println!("Listener recevied flags: {}", event.flags); if event.flags & syscall::EVENT_WRITE == syscall::EVENT_WRITE { - let stream = syscall::dup(server.as_raw_fd() as usize, b"listen").map_err(from_syscall_error)?; - let stream = unsafe { File::from_raw_fd(stream as RawFd) }; + loop { + let stream = match syscall::dup(server.as_raw_fd() as usize, b"listen").map_err(from_syscall_error) { + Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break, + stream => stream? + }; + let stream = unsafe { File::from_raw_fd(stream as RawFd) }; - nonblock(&stream)?; - event_file.write(&syscall::Event { - id: stream.as_raw_fd() as usize, - data: next_id, - flags: syscall::EVENT_READ | syscall::EVENT_WRITE, - })?; + event_file.write(&syscall::Event { + id: stream.as_raw_fd() as usize, + data: next_id, + flags: syscall::EVENT_READ | syscall::EVENT_WRITE, + })?; - clients.insert(next_id, stream); - next_id += 1; + clients.insert(next_id, stream); + println!("-> Spawned client #{}", next_id); + next_id += 1; + } } } else { println!("Client #{} received flags: {}", event.data, event.flags); let client = clients.get_mut(&event.data).unwrap(); - if event.flags & syscall::EVENT_WRITE == syscall::EVENT_WRITE { - println!("-> Writing"); - const BUF: &str = "Hello from the log server\n"; - let mut written = 0; - while written < BUF.len() { - let len = match client.write(BUF[written..].as_bytes()) { - Ok(0) => { - println!("--> EOF"); - clients.remove(&event.data); - continue 'outer; - }, - Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break, - len => len? - }; - println!("--> Wrote {}/{} bytes: {:?}", len, BUF.len(), &BUF[written..]); - written += len; - } - } if event.flags & syscall::EVENT_READ == syscall::EVENT_READ { println!("-> Reading"); let mut buf = [0; 128]; @@ -84,7 +68,21 @@ fn main() -> io::Result<()> { Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break, len => len? }; - println!("--> Read {}/128 bytes: {:?}", len, str::from_utf8(&buf)); + println!("--> Read {}/128 bytes: {:?}", len, str::from_utf8(&buf[..len])); + } + } + if event.flags & syscall::EVENT_WRITE == syscall::EVENT_WRITE { + println!("-> Writing"); + const BUF: &str = "Hello from the log server\n"; + let mut written = 0; + while written < BUF.len() { + let len = match client.write(BUF[written..].as_bytes()) { + Ok(0) => panic!("EOF should never happen here"), + Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break, + len => len? + }; + println!("--> Wrote {}/{} bytes: {:?}", len, BUF.len(), &BUF[written..]); + written += len; } } } diff --git a/examples/main.rs b/examples/main.rs index 2a41aebb6d..7a15762386 100644 --- a/examples/main.rs +++ b/examples/main.rs @@ -1,5 +1,5 @@ use std::{ - fs::File, + fs::{File, OpenOptions}, io::{self, prelude::*}, os::unix::io::{AsRawFd, FromRawFd, RawFd}, thread, @@ -23,11 +23,24 @@ fn main() -> io::Result<()> { let mut buf = [0; 5]; let server = File::create("chan:hello_world")?; { - let mut client = File::open("chan:hello_world")?; - // First client not accepted yet - assert_eq!(File::open("chan:hello_world").unwrap_err().kind(), io::ErrorKind::ConnectionRefused); + println!("Testing O_EXCL..."); + assert_eq!( + OpenOptions::new() + .write(true) + .create_new(true) + .open("chan:hello_world").unwrap_err().kind(), + io::ErrorKind::AlreadyExists + ); + + println!("Testing connecting..."); + + File::open("chan:hello_world")?; // closed connection will silently be skipped + let mut client = File::create("chan:hello_world")?; // O_CREAT without O_EXCL does nothing + let tmp = File::open("chan:hello_world")?; // multiple connections are handled let mut stream = dup(&server, "listen")?; + assert!(dup(&server, "listen").is_ok()); + drop(tmp); println!("Testing basic I/O..."); @@ -89,9 +102,6 @@ fn main() -> io::Result<()> { println!("-> Write before accept would block"); } - assert_eq!(dup(&server, "listen").unwrap_err().kind(), io::ErrorKind::ConnectionReset); - println!("-> Server can't accept dropped client"); - let mut client = dup(&server, "connect")?; nonblock(&client)?; diff --git a/src/chan.rs b/src/chan.rs index 7ce69b3d95..0378d0099a 100644 --- a/src/chan.rs +++ b/src/chan.rs @@ -1,20 +1,21 @@ use crate::post_fevent; use std::{ cmp, - collections::HashMap, + collections::{HashMap, VecDeque}, fs::File, - io, - mem + io }; use syscall::{flag::*, error::*, Error, SchemeBlockMut, Result}; #[derive(Debug, Default)] pub struct Client { - buffer: Vec + buffer: Vec, + remote: Connection } #[derive(Debug, Default)] pub struct Listener { - path: Option + path: Option, + awaiting: VecDeque } #[derive(Debug)] pub enum Extra { @@ -42,43 +43,48 @@ impl Default for Connection { pub struct Handle { id: usize, flags: usize, - remote: Connection, extra: Extra } impl Handle { - /// Duplicate this listener handle (no checks done) into one that - /// can be connected to. - pub fn accept(&mut self) -> Self { + /// Duplicate this listener handle into one that is linked to the + /// specified remote. + /// Does NOT error if this is not a listener + pub fn accept(&self, remote: usize) -> Self { Self { flags: self.flags, - remote: mem::replace(&mut self.remote, Connection::Waiting), + extra: Extra::Client(Client { + remote: Connection::Open(remote), + ..Client::default() + }), ..Default::default() } } - /// Mark this listener handle (no checks done) as having a - /// connection which can be accepted, but only if it is ready to - /// accept. + /// Mark this listener handle as having a connection which can be + /// accepted, but only if it is ready to accept. + /// Errors if this is not a listener pub fn connect(&mut self, other: usize) -> Result<()> { - if self.remote != Connection::Waiting { - return Err(Error::new(ECONNREFUSED)); + match self.extra { + Extra::Listener(ref mut listener) => { + listener.awaiting.push_back(other); + Ok(()) + }, + _ => Err(Error::new(EBADF)) } - self.remote = Connection::Open(other); - Ok(()) } /// Error if this is not a listener - pub fn require_listener(&self) -> Result<()> { + pub fn require_listener(&mut self) -> Result<&mut Listener> { match self.extra { - Extra::Listener(_) => Ok(()), + Extra::Listener(ref mut listener) => Ok(listener), _ => Err(Error::new(EBADF)) } } /// Error if this is not a client - pub fn require_client(&self) -> Result<()> { + pub fn require_client(&mut self) -> Result<&mut Client> { match self.extra { - Extra::Client(_) => Ok(()), + Extra::Client(ref mut client) => Ok(client), _ => Err(Error::new(EBADF)) } } @@ -102,6 +108,12 @@ impl ChanScheme { } impl SchemeBlockMut for ChanScheme { + // ___ ____ _____ _ _ + // / _ \| _ \| ____| \ | | + // | | | | |_) | _| | \| | + // | |_| | __/| |___| |\ | + // \___/|_| |_____|_| \_| + fn open(&mut self, path: &[u8], flags: usize, _uid: u32, _gid: u32) -> Result> { let path = ::std::str::from_utf8(path).or(Err(Error::new(EPERM)))?; @@ -109,16 +121,17 @@ impl SchemeBlockMut for ChanScheme { let mut new = Handle::default(); new.flags = flags; - if flags & O_CREAT == O_CREAT { - if self.listeners.contains_key(path) { - return Err(Error::new(EADDRINUSE)); - } + let create = flags & O_CREAT == O_CREAT; + + if create && !self.listeners.contains_key(path) { let mut listener = Listener::default(); if !path.is_empty() { self.listeners.insert(String::from(path), new_id); listener.path = Some(String::from(path)); } new.extra = Extra::Listener(listener); + } else if create && flags & O_EXCL == O_EXCL { + return Err(Error::new(EEXIST)); } else { let listener_id = *self.listeners.get(path).ok_or(Error::new(ENOENT))?; let listener = self.handles.get_mut(&listener_id).expect("orphan listener left over"); @@ -137,29 +150,37 @@ impl SchemeBlockMut for ChanScheme { fn dup(&mut self, id: usize, buf: &[u8]) -> Result> { match buf { b"listen" => { - let (flags, remote) = { - let handle = self.handles.get(&id).ok_or(Error::new(EBADF))?; - handle.require_listener()?; - (handle.flags, handle.remote) - }; - if let Connection::Open(remote_id) = remote { - let new_id = self.next_id; - let new = self.handles.get_mut(&id).map(Handle::accept).unwrap(); + loop { + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + let listener = handle.require_listener()?; - // Hook the remote side, assuming it's still - // connected, up to this one so the connection is - // mutal. - let mut remote = self.handles.get_mut(&remote_id).ok_or(Error::new(ECONNRESET))?; - remote.remote = Connection::Open(new_id); - post_fevent(&mut self.socket, remote_id, EVENT_WRITE)?; + break if let Some(remote_id) = listener.awaiting.pop_front() { + let new_id = self.next_id; + let new = handle.accept(remote_id); - self.handles.insert(new_id, new); - self.next_id += 1; - Ok(Some(new_id)) - } else if flags & O_NONBLOCK == O_NONBLOCK { - Err(Error::new(EAGAIN)) - } else { - Ok(None) + // Hook the remote side, assuming it's still + // connected, up to this one so the connection is + // mutal. + let remote = match self.handles.get_mut(&remote_id) { + Some(client) => client, + None => continue // Check next client + }; + match remote.extra { + Extra::Client(ref mut client) => { + client.remote = Connection::Open(new_id); + }, + Extra::Listener(_) => panic!("newly created handle can't possibly be a listener") + } + post_fevent(&mut self.socket, remote_id, EVENT_WRITE)?; + + self.handles.insert(new_id, new); + self.next_id += 1; + Ok(Some(new_id)) + } else if handle.flags & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Ok(None) + }; } }, b"connect" => { @@ -184,6 +205,95 @@ impl SchemeBlockMut for ChanScheme { } } } + + // ___ ___ ___ ____ _ ___ ____ _____ + // |_ _/ _ \ ( _ ) / ___| | / _ \/ ___|| ____| + // | | | | | / _ \/\ | | | | | | | \___ \| _| + // | | |_| | | (_> < | |___| |__| |_| |___) | |___ + // |___\___/ \___/\/ \____|_____\___/|____/|_____| + + fn write(&mut self, id: usize, buf: &[u8]) -> Result> { + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + let client = handle.require_client()?; + + if let Connection::Open(remote_id) = client.remote { + let remote = self.handles.get_mut(&remote_id).unwrap(); + match remote.extra { + Extra::Client(ref mut client) => { + client.buffer.extend(buf); + if client.buffer.len() == buf.len() { + // Send readable only if it wasn't readable + // before + post_fevent(&mut self.socket, remote_id, EVENT_READ)?; + } + Ok(Some(buf.len())) + }, + Extra::Listener(_) => panic!("somehow, a client was connected to a listener directly") + } + } else if client.remote == Connection::Closed { + Err(Error::new(EPIPE)) + } else if handle.flags & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Ok(None) + } + } + fn fsync(&mut self, id: usize) -> Result> { + self.handles.get(&id) + .ok_or(Error::new(EBADF)) + .and(Ok(Some(id))) + } + fn read(&mut self, id: usize, buf: &mut [u8]) -> Result> { + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + let client = handle.require_client()?; + + if !client.buffer.is_empty() { + let len = cmp::min(buf.len(), client.buffer.len()); + buf[..len].copy_from_slice(&client.buffer[..len]); + client.buffer.drain(..len); + Ok(Some(len)) + } else if client.remote == Connection::Closed { + // Remote dropped, send EOF + Ok(Some(0)) + } else if handle.flags & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Ok(None) + } + } + fn close(&mut self, id: usize) -> Result> { + let handle = self.handles.remove(&id).ok_or(Error::new(EBADF))?; + + match handle.extra { + Extra::Client(client) => if let Connection::Open(remote_id) = client.remote { + let remote = self.handles.get_mut(&remote_id).unwrap(); + + match remote.extra { + Extra::Client(ref mut client) => { + client.remote = Connection::Closed; + if client.buffer.is_empty() { + // Post readable on EOF only if it wasn't + // readable before + post_fevent(&mut self.socket, remote_id, EVENT_READ)?; + } + }, + Extra::Listener(_) => panic!("a client can't be connected to a listener!") + } + }, + Extra::Listener(listener) => if let Some(path) = listener.path { + self.listeners.remove(&path); + } + } + Ok(Some(0)) + } + + + // ____ _ ____ _ __ __ _____ _____ _____ ____ ____ + // | _ \ / \ | _ \ / \ | \/ | ____|_ _| ____| _ \/ ___| + // | |_) / _ \ | |_) | / _ \ | |\/| | _| | | | _| | |_) \___ \ + // | __/ ___ \| _ < / ___ \| | | | |___ | | | |___| _ < ___) | + // |_| /_/ \_\_| \_\/_/ \_\_| |_|_____| |_| |_____|_| \_\____/ + fn fcntl(&mut self, id: usize, cmd: usize, arg: usize) -> Result> { let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; match cmd { @@ -200,101 +310,17 @@ impl SchemeBlockMut for ChanScheme { let mut events = 0; match handle.extra { Extra::Client(ref client) => { - if let Connection::Open(_) = handle.remote { + if let Connection::Open(_) = client.remote { events |= EVENT_WRITE; } - if !client.buffer.is_empty() || handle.remote == Connection::Closed { + if !client.buffer.is_empty() || client.remote == Connection::Closed { events |= EVENT_READ; } }, - Extra::Listener(_) => if let Connection::Open(_) = handle.remote { - events |= EVENT_READ; + Extra::Listener(ref listener) => if !listener.awaiting.is_empty() { + events |= EVENT_READ | EVENT_WRITE; } } Ok(Some(events)) } - fn write(&mut self, id: usize, buf: &[u8]) -> Result> { - let (flags, remote) = { - let handle = self.handles.get(&id).ok_or(Error::new(EBADF))?; - handle.require_client()?; - (handle.flags, handle.remote) - }; - if let Connection::Open(remote_id) = remote { - let remote = self.handles.get_mut(&remote_id).unwrap(); - match remote.extra { - Extra::Client(ref mut client) => { - client.buffer.extend(buf); - if client.buffer.len() == buf.len() { - // Send readable only if it wasn't readable - // before - post_fevent(&mut self.socket, remote_id, EVENT_READ)?; - } - Ok(Some(buf.len())) - }, - Extra::Listener(_) => panic!("somehow, a client was connected to a listener directly") - } - } else if remote == Connection::Closed { - Err(Error::new(EPIPE)) - } else if flags & O_NONBLOCK == O_NONBLOCK { - Err(Error::new(EAGAIN)) - } else { - Ok(None) - } - } - fn fsync(&mut self, id: usize) -> Result> { - self.handles.get(&id) - .ok_or(Error::new(EBADF)) - .and(Ok(Some(id))) - } - fn read(&mut self, id: usize, buf: &mut [u8]) -> Result> { - let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; - - let client = match handle.extra { - Extra::Client(ref mut client) => client, - Extra::Listener(_) => return Err(Error::new(EBADF)) - }; - - if !client.buffer.is_empty() { - let len = cmp::min(buf.len(), client.buffer.len()); - buf[..len].copy_from_slice(&client.buffer[..len]); - client.buffer.drain(..len); - Ok(Some(len)) - } else if handle.remote == Connection::Closed { - // Remote dropped, send EOF - Ok(Some(0)) - } else if handle.flags & O_NONBLOCK == O_NONBLOCK { - Err(Error::new(EAGAIN)) - } else { - Ok(None) - } - } - fn close(&mut self, id: usize) -> Result> { - let handle = self.handles.remove(&id).ok_or(Error::new(EBADF))?; - - match handle.extra { - Extra::Client(_) => { - if let Connection::Open(remote_id) = handle.remote { - let mut remote = self.handles.get_mut(&remote_id).unwrap(); - remote.remote = Connection::Closed; - match remote.extra { - Extra::Client(ref client) => if client.buffer.is_empty() { - // Post readable on EOF only if it wasn't - // readable before - post_fevent(&mut self.socket, remote_id, EVENT_READ)?; - }, - Extra::Listener(_) => panic!("a client can't be connected to a listener!") - } - } - }, - Extra::Listener(listener) => { - // Clients never reference listeners in any way, it's - // safe to drop - - if let Some(path) = listener.path { - self.listeners.remove(&path); - } - } - } - Ok(Some(0)) - } }