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

Support for tcp listen.

Browse Source
This commit is contained in:
Egor Karavaev
2017-10-04 18:43:07 +03:00
parent 5b018426d2
commit 454dfd9a39
7 changed files with 227 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/smolnetd/buffer_pool.rs
+1 -3
View File
@@ -68,9 +68,7 @@ impl BufferPool {
pub fn get_buffer(&mut self) -> Buffer {
let buffer = match self.stack.borrow_mut().pop() {
None => {
vec![0u8; self.buffers_size]
}
None => vec![0u8; self.buffers_size],
Some(mut v) => {
// memsetting the buffer with `resize` would be a waste of time
let capacity = v.capacity();
src/smolnetd/main.rs
+2
View File
@@ -1,3 +1,5 @@
#![feature(drain_filter)]
extern crate event;
#[macro_use]
extern crate log;
src/smolnetd/scheme/ip.rs
+30 -11
View File
@@ -161,19 +161,38 @@ impl<'a, 'b> SchemeSocket for RawSocket<'a, 'b> {
socket_handle: SocketHandle,
path: &str,
_: &mut Self::SchemeDataT,
) -> syscall::Result<SchemeFile<Self>> {
) -> syscall::Result<DupResult<Self>> {
match path {
"write_timeout" => Ok(SchemeFile::Setting(SettingFile {
socket_handle,
fd,
setting: Setting::WriteTimeout,
})),
"read_timeout" => Ok(SchemeFile::Setting(SettingFile {
socket_handle,
fd,
setting: Setting::ReadTimeout,
})),
"write_timeout" => Ok((
SchemeFile::Setting(SettingFile {
socket_handle,
fd,
setting: Setting::WriteTimeout,
}),
None,
)),
"read_timeout" => Ok((
SchemeFile::Setting(SettingFile {
socket_handle,
fd,
setting: Setting::ReadTimeout,
}),
None,
)),
_ => Err(syscall::Error::new(syscall::EBADF)),
}
}
fn fpath(&self, _file: &SchemeFile<Self>, buf: &mut [u8]) -> syscall::Result<usize> {
let path = format!("ip:{}", self.ip_protocol());
let path = path.as_bytes();
let mut i = 0;
while i < buf.len() && i < path.len() {
buf[i] = path[i];
i += 1;
}
Ok(i)
}
}
src/smolnetd/scheme/mod.rs
-1
View File
@@ -181,7 +181,6 @@ impl Smolnetd {
if count == 0 {
break;
}
trace!("got frame {}", count);
buffer.resize(count);
self.input_queue.borrow_mut().push_back(buffer);
total_frames += 1;
src/smolnetd/scheme/socket.rs
+109 -66
View File
@@ -29,7 +29,7 @@ impl<DataT> SocketFile<DataT> {
read_timeout: self.read_timeout,
write_timeout: self.write_timeout,
socket_handle: self.socket_handle,
data
data,
}
}
@@ -40,7 +40,7 @@ impl<DataT> SocketFile<DataT> {
read_timeout: None,
write_timeout: None,
socket_handle,
data
data,
}
}
}
@@ -77,11 +77,11 @@ struct WaitHandle {
packet: syscall::Packet,
}
#[derive(Default)]
struct WaitQueues {
read_queue: Vec<WaitHandle>,
write_queue: Vec<WaitHandle>,
}
type WaitQueue = Vec<WaitHandle>;
type WaitQueueMap = BTreeMap<SocketHandle, WaitQueue>;
pub type DupResult<T: SchemeSocket> = (SchemeFile<T>, Option<(Socket<'static, 'static>, T::DataT)>);
pub trait SchemeSocket
where
@@ -109,6 +109,8 @@ where
fn write_buf(&mut self, &mut SocketFile<Self::DataT>, buf: &[u8]) -> syscall::Result<usize>;
fn read_buf(&mut self, &mut SocketFile<Self::DataT>, buf: &mut [u8]) -> syscall::Result<usize>;
fn fpath(&self, &SchemeFile<Self>, &mut [u8]) -> syscall::Result<usize>;
fn dup(
&self,
&mut SchemeFile<Self>,
@@ -116,11 +118,9 @@ where
SocketHandle,
&str,
&mut Self::SchemeDataT,
) -> syscall::Result<SchemeFile<Self>>;
) -> syscall::Result<DupResult<Self>>;
}
type WaitQueueMap = BTreeMap<SocketHandle, WaitQueues>;
pub struct SocketScheme<SocketT>
where
SocketT: SchemeSocket,
@@ -206,35 +206,19 @@ where
let socket_handles: Vec<_> = self.wait_queue_map.keys().cloned().collect();
for socket_handle in socket_handles {
let (can_recv, can_send) = {
let mut socket_set = self.socket_set.borrow_mut();
let socket: &mut SocketT = socket_set.get_mut(socket_handle).as_socket();
(socket.can_recv(), socket.can_send())
};
if can_recv {
self.wake_up_wait_queue(socket_handle, cur_time, |wq| &mut wq.read_queue)?;
}
if can_send {
self.wake_up_wait_queue(socket_handle, cur_time, |wq| &mut wq.write_queue)?;
}
self.wake_up_wait_queue(socket_handle, cur_time)?;
}
Ok(())
}
fn wake_up_wait_queue<F>(
fn wake_up_wait_queue(
&mut self,
socket_handle: SocketHandle,
cur_time: syscall::TimeSpec,
f: F,
) -> Result<()>
where
F: Fn(&mut WaitQueues) -> &mut Vec<WaitHandle>,
{
let mut input_queue = if let Some(wait_queues) = self.wait_queue_map.get_mut(&socket_handle)
) -> Result<()> {
let mut input_queue = if let Some(wait_queue) = self.wait_queue_map.get_mut(&socket_handle)
{
::std::mem::replace(f(wait_queues), vec![])
::std::mem::replace(wait_queue, vec![])
} else {
vec![]
};
@@ -247,11 +231,10 @@ where
if packet.a == (-syscall::EWOULDBLOCK) as usize {
match wait_handle.until {
Some(until)
if (until.tv_sec >= cur_time.tv_sec
if (until.tv_sec < cur_time.tv_sec
|| (until.tv_sec == cur_time.tv_sec
&& until.tv_nsec >= cur_time.tv_nsec)) =>
&& until.tv_nsec < cur_time.tv_nsec)) =>
{
trace!("Timeouting fd {}", packet.b);
packet.a = (-syscall::ETIMEDOUT) as usize;
self.scheme_file.write_all(&packet)?;
}
@@ -260,20 +243,41 @@ where
}
}
} else {
trace!("Waking up fd {}", packet.b);
trace!("Waking up {} with {}", packet.b, packet.a);
self.scheme_file.write_all(&packet)?;
}
}
if let Some(wait_queues) = self.wait_queue_map.get_mut(&socket_handle) {
f(wait_queues).extend(to_retain);
if let Some(wait_queue) = self.wait_queue_map.get_mut(&socket_handle) {
wait_queue.extend(to_retain);
}
Ok(())
}
fn move_handle_to_new_socket_handle(
wait_queue_map: &mut WaitQueueMap,
fd: usize,
from: SocketHandle,
to: SocketHandle,
) -> syscall::Result<()> {
trace!("Moving {} from {:?} to {:?}", fd, from, to);
let mut to_move = vec![];
if let Some(wait_queue) = wait_queue_map.get_mut(&from) {
to_move = wait_queue
.drain_filter(|wh| wh.packet.b == fd)
.collect::<Vec<_>>();
}
wait_queue_map
.entry(to)
.or_insert_with(|| vec![])
.extend(to_move);
Ok(())
}
fn handle_block(&mut self, mut packet: syscall::Packet) -> Result<()> {
trace!("Handling blocking call");
let syscall_result = self.try_handle_block(&mut packet);
if let Err(syscall_error) = syscall_result {
packet.a = (-syscall_error.errno) as usize;
@@ -306,9 +310,9 @@ where
}?;
let mut timeout = match packet.a {
syscall::SYS_READ => Ok(read_timeout),
syscall::SYS_WRITE => Ok(write_timeout),
_ => Err(syscall::Error::new(syscall::EBADF)),
syscall::SYS_READ => Ok(read_timeout),
_ => Ok(None),
}?;
if let Some(ref mut timeout) = timeout {
@@ -317,18 +321,11 @@ where
*timeout = add_time(timeout, &cur_time)
}
trace!("Adding {} to wait queie", fd);
let wait_queues = self.wait_queue_map
let wait_queue = self.wait_queue_map
.entry(socket_handle)
.or_insert_with(WaitQueues::default);
.or_insert_with(|| vec![]);
let queue = match packet.a {
syscall::SYS_READ => Ok(&mut wait_queues.read_queue),
syscall::SYS_WRITE => Ok(&mut wait_queues.write_queue),
_ => Err(syscall::Error::new(syscall::EBADF)),
}?;
queue.push(WaitHandle {
wait_queue.push(WaitHandle {
until: timeout,
packet: *packet,
});
@@ -381,6 +378,7 @@ where
{
fn open(&mut self, url: &[u8], flags: usize, uid: u32, _gid: u32) -> syscall::Result<usize> {
let path = str::from_utf8(url).or_else(|_| Err(syscall::Error::new(syscall::EINVAL)))?;
trace!("open {}", path);
let (socket, data) = SocketT::new_socket(path, uid, &mut self.scheme_data)?;
@@ -403,6 +401,7 @@ where
}
fn close(&mut self, fd: usize) -> syscall::Result<usize> {
trace!("close {}", fd);
let socket_handle = {
let handle = self.fds
.get(&fd)
@@ -416,20 +415,14 @@ where
socket.close_file(&scheme_file, &mut self.scheme_data)?;
}
let remove_wq =
if let Some(ref mut wait_queues) = self.wait_queue_map.get_mut(&socket_handle) {
wait_queues.read_queue.retain(
if let Some(ref mut wait_queue) = self.wait_queue_map.get_mut(&socket_handle) {
wait_queue.retain(
|&WaitHandle {
packet: syscall::Packet { a, .. },
..
}| a != fd,
);
wait_queues.write_queue.retain(
|&WaitHandle {
packet: syscall::Packet { a, .. },
..
}| a != fd,
);
wait_queues.read_queue.is_empty() && wait_queues.write_queue.is_empty()
wait_queue.is_empty()
} else {
false
};
@@ -485,22 +478,42 @@ where
fn dup(&mut self, fd: usize, buf: &[u8]) -> syscall::Result<usize> {
let path = str::from_utf8(buf).or_else(|_| Err(syscall::Error::new(syscall::EINVAL)))?;
let handle = {
let new_file = {
let handle = self.fds
.get_mut(&fd)
.ok_or_else(|| syscall::Error::new(syscall::EBADF))?;
let socket_handle = handle.socket_handle();
let mut socket_set = self.socket_set.borrow_mut();
let socket: &mut SocketT = socket_set.get_mut(handle.socket_handle()).as_socket();
socket.dup(handle, fd, socket_handle, path, &mut self.scheme_data)
}?;
let (new_handle, update_with) = {
let mut socket_set = self.socket_set.borrow_mut();
let socket: &mut SocketT = socket_set.get_mut(handle.socket_handle()).as_socket();
socket.dup(handle, fd, socket_handle, path, &mut self.scheme_data)?
};
self.socket_set.borrow_mut().retain(handle.socket_handle());
if let Some((socket, data)) = update_with {
if let SchemeFile::Socket(ref mut handle) = *handle {
let socket_handle = self.socket_set.borrow_mut().add(socket);
trace!("Updating {} to a new socket handle {:?}", fd, socket_handle);
Self::move_handle_to_new_socket_handle(
&mut self.wait_queue_map,
fd,
handle.socket_handle,
socket_handle,
)?;
handle.socket_handle = socket_handle;
handle.data = data;
} else {
self.socket_set.borrow_mut().retain(handle.socket_handle());
}
} else {
self.socket_set.borrow_mut().retain(handle.socket_handle());
}
new_handle
};
let id = self.next_fd;
self.fds.insert(id, handle);
self.fds.insert(id, new_file);
self.next_fd += 1;
Ok(id)
@@ -529,6 +542,36 @@ where
// TODO Implement fsyncing
// self.0.network_fsync()
}
fn fpath(&mut self, fd: usize, buf: &mut [u8]) -> syscall::Result<usize> {
let handle = self.fds
.get_mut(&fd)
.ok_or_else(|| syscall::Error::new(syscall::EBADF))?;
let mut socket_set = self.socket_set.borrow_mut();
let socket: &mut SocketT = socket_set.get_mut(handle.socket_handle()).as_socket();
socket.fpath(handle, buf)
}
fn fcntl(&mut self, fd: usize, cmd: usize, arg: usize) -> syscall::Result<usize> {
let handle = self.fds
.get_mut(&fd)
.ok_or_else(|| syscall::Error::new(syscall::EBADF))?;
if let SchemeFile::Socket(ref mut socket_file) = *handle {
match cmd {
syscall::F_GETFL => Ok(socket_file.flags),
syscall::F_SETFL => {
socket_file.flags = arg & !syscall::O_ACCMODE;
Ok(0)
}
_ => Err(syscall::Error::new(syscall::EINVAL)),
}
} else {
Err(syscall::Error::new(syscall::EBADF))
}
}
}
fn add_time(a: &TimeSpec, b: &TimeSpec) -> TimeSpec {
src/smolnetd/scheme/tcp.rs
+65 -18
View File
@@ -116,7 +116,7 @@ impl<'a> SchemeSocket for TcpSocket<'a> {
let tx_packets = vec![0; 65_535];
let rx_buffer = TcpSocketBuffer::new(rx_packets);
let tx_buffer = TcpSocketBuffer::new(tx_packets);
let mut tcp_socket = TcpSocket::new(rx_buffer, tx_buffer);
let mut socket = TcpSocket::new(rx_buffer, tx_buffer);
if local_endpoint.port == 0 {
local_endpoint.port = port_set
@@ -127,14 +127,22 @@ impl<'a> SchemeSocket for TcpSocket<'a> {
}
{
let tcp_socket: &mut TcpSocket = tcp_socket.as_socket();
trace!("Connecting tcp {} {}", local_endpoint, remote_endpoint);
tcp_socket
.connect(remote_endpoint, local_endpoint)
.expect("Can't connect tcp socket ");
let tcp_socket: &mut TcpSocket = socket.as_socket();
if remote_endpoint.is_specified() {
trace!("Connecting tcp {} {}", local_endpoint, remote_endpoint);
tcp_socket
.connect(remote_endpoint, local_endpoint)
.expect("Can't connect tcp socket ");
} else {
trace!("Listening tcp {}", local_endpoint);
tcp_socket
.listen(local_endpoint)
.expect("Can't listen on local endpoint");
}
}
Ok((tcp_socket, ()))
Ok((socket, ()))
}
fn close_file(
@@ -153,10 +161,9 @@ impl<'a> SchemeSocket for TcpSocket<'a> {
file: &mut SocketFile<Self::DataT>,
buf: &[u8],
) -> syscall::Result<usize> {
if !self.is_open() {
return Err(syscall::Error::new(syscall::EADDRNOTAVAIL));
}
if self.can_send() {
if !self.is_active() {
Err(syscall::Error::new(syscall::ENOTCONN))
} else if self.can_send() {
self.send_slice(buf).expect("Can't send slice");
Ok(buf.len())
} else if file.flags & syscall::O_NONBLOCK == syscall::O_NONBLOCK {
@@ -171,7 +178,9 @@ impl<'a> SchemeSocket for TcpSocket<'a> {
file: &mut SocketFile<Self::DataT>,
buf: &mut [u8],
) -> syscall::Result<usize> {
if self.can_recv() {
if !self.is_active() {
Err(syscall::Error::new(syscall::ENOTCONN))
} else if self.can_recv() {
let length = self.recv_slice(buf).expect("Can't receive slice");
Ok(length)
} else if file.flags & syscall::O_NONBLOCK == syscall::O_NONBLOCK {
@@ -188,8 +197,7 @@ impl<'a> SchemeSocket for TcpSocket<'a> {
socket_handle: SocketHandle,
path: &str,
port_set: &mut Self::SchemeDataT,
) -> syscall::Result<SchemeFile<Self>> {
trace!("TCP dup {}", path);
) -> syscall::Result<DupResult<Self>> {
let handle = match path {
"ttl" => SchemeFile::Setting(SettingFile {
socket_handle,
@@ -206,18 +214,57 @@ impl<'a> SchemeSocket for TcpSocket<'a> {
fd,
setting: Setting::WriteTimeout,
}),
_ => if let SchemeFile::Socket(ref tcp_handle) = *file {
SchemeFile::Socket(tcp_handle.clone_with_data(()))
"listen" => if let SchemeFile::Socket(ref tcp_handle) = *file {
if !self.is_active() {
return Err(syscall::Error::new(syscall::EWOULDBLOCK));
}
trace!("TCP creating new listening socket");
let new_handle = SchemeFile::Socket(tcp_handle.clone_with_data(()));
let rx_packets = vec![0; 65_535];
let tx_packets = vec![0; 65_535];
let rx_buffer = TcpSocketBuffer::new(rx_packets);
let tx_buffer = TcpSocketBuffer::new(tx_packets);
let mut socket = TcpSocket::new(rx_buffer, tx_buffer);
{
let tcp_socket: &mut TcpSocket = socket.as_socket();
tcp_socket
.listen(self.local_endpoint())
.expect("Can't listen on local endpoint");
}
port_set.acquire_port(self.local_endpoint().port);
return Ok((new_handle, Some((socket, ()))));
} else {
SchemeFile::Socket(SocketFile::new_with_data(socket_handle, ()))
return Err(syscall::Error::new(syscall::EBADF));
},
_ => {
trace!("TCP dup unknown {}", path);
if let SchemeFile::Socket(ref tcp_handle) = *file {
SchemeFile::Socket(tcp_handle.clone_with_data(()))
} else {
SchemeFile::Socket(SocketFile::new_with_data(socket_handle, ()))
}
}
};
if let SchemeFile::Socket(_) = handle {
port_set.acquire_port(self.local_endpoint().port);
}
Ok(handle)
Ok((handle, None))
}
fn fpath(&self, _: &SchemeFile<Self>, buf: &mut [u8]) -> syscall::Result<usize> {
let path = format!("tcp:{}/{}", self.remote_endpoint(), self.local_endpoint());
let path = path.as_bytes();
let mut i = 0;
while i < buf.len() && i < path.len() {
buf[i] = path[i];
i += 1;
}
Ok(i)
}
}
src/smolnetd/scheme/udp.rs
+20 -2
View File
@@ -82,6 +82,7 @@ impl<'a, 'b> SchemeSocket for UdpSocket<'a, 'b> {
(Some(timespec), count)
}
};
trace!("Setting {:?} to {:?}", setting, timeout);
match setting {
Setting::ReadTimeout => {
file.read_timeout = timeout;
@@ -190,7 +191,7 @@ impl<'a, 'b> SchemeSocket for UdpSocket<'a, 'b> {
socket_handle: SocketHandle,
path: &str,
port_set: &mut Self::SchemeDataT,
) -> syscall::Result<SchemeFile<Self>> {
) -> syscall::Result<DupResult<Self>> {
let handle = match path {
"ttl" => SchemeFile::Setting(SettingFile {
socket_handle,
@@ -227,7 +228,24 @@ impl<'a, 'b> SchemeSocket for UdpSocket<'a, 'b> {
port_set.acquire_port(self.endpoint().port);
}
Ok(handle)
Ok((handle, None))
}
fn fpath(&self, file: &SchemeFile<Self>, buf: &mut [u8]) -> syscall::Result<usize> {
if let &SchemeFile::Socket(ref socket_file) = file {
let path = format!("udp:{}/{}", socket_file.data, self.endpoint());
let path = path.as_bytes();
let mut i = 0;
while i < buf.len() && i < path.len() {
buf[i] = path[i];
i += 1;
}
Ok(i)
} else {
Err(syscall::Error::new(syscall::EBADF))
}
}
}