vasilito/RedBear-OS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Support multiple connectors to listener

Browse Source
This commit is contained in:
jD91mZM2
2019-04-25 17:06:37 +02:00
parent 50e0b499f4
commit 85f6cea0bd
3 changed files with 210 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/logserver.rs
+33 -35
View File
@@ -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;
}
}
}
examples/main.rs
+17 -7
View File
@@ -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)?;
src/chan.rs
+160 -134
View File
@@ -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<u8>
buffer: Vec<u8>,
remote: Connection
}
#[derive(Debug, Default)]
pub struct Listener {
path: Option<String>
path: Option<String>,
awaiting: VecDeque<usize>
}
#[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<Option<usize>> {
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<Option<usize>> {
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<Option<usize>> {
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<Option<usize>> {
self.handles.get(&id)
.ok_or(Error::new(EBADF))
.and(Ok(Some(id)))
}
fn read(&mut self, id: usize, buf: &mut [u8]) -> Result<Option<usize>> {
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<Option<usize>> {
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<Option<usize>> {
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<Option<usize>> {
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<Option<usize>> {
self.handles.get(&id)
.ok_or(Error::new(EBADF))
.and(Ok(Some(id)))
}
fn read(&mut self, id: usize, buf: &mut [u8]) -> Result<Option<usize>> {
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<Option<usize>> {
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))
}
}