From 454dfd9a39c59f357f2de59a834781c94040748e Mon Sep 17 00:00:00 2001 From: Egor Karavaev Date: Wed, 4 Oct 2017 18:43:07 +0300 Subject: [PATCH] Support for tcp listen. --- src/smolnetd/buffer_pool.rs | 4 +- src/smolnetd/main.rs | 2 + src/smolnetd/scheme/ip.rs | 41 +++++--- src/smolnetd/scheme/mod.rs | 1 - src/smolnetd/scheme/socket.rs | 175 +++++++++++++++++++++------------- src/smolnetd/scheme/tcp.rs | 83 ++++++++++++---- src/smolnetd/scheme/udp.rs | 22 ++++- 7 files changed, 227 insertions(+), 101 deletions(-) diff --git a/src/smolnetd/buffer_pool.rs b/src/smolnetd/buffer_pool.rs index f3248c1958..c96baea781 100644 --- a/src/smolnetd/buffer_pool.rs +++ b/src/smolnetd/buffer_pool.rs @@ -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(); diff --git a/src/smolnetd/main.rs b/src/smolnetd/main.rs index 2ac8cc8095..1f78c09590 100644 --- a/src/smolnetd/main.rs +++ b/src/smolnetd/main.rs @@ -1,3 +1,5 @@ +#![feature(drain_filter)] + extern crate event; #[macro_use] extern crate log; diff --git a/src/smolnetd/scheme/ip.rs b/src/smolnetd/scheme/ip.rs index 6c8c74a2c8..d48fd52f4f 100644 --- a/src/smolnetd/scheme/ip.rs +++ b/src/smolnetd/scheme/ip.rs @@ -161,19 +161,38 @@ impl<'a, 'b> SchemeSocket for RawSocket<'a, 'b> { socket_handle: SocketHandle, path: &str, _: &mut Self::SchemeDataT, - ) -> syscall::Result> { + ) -> syscall::Result> { 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, buf: &mut [u8]) -> syscall::Result { + 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) + } } diff --git a/src/smolnetd/scheme/mod.rs b/src/smolnetd/scheme/mod.rs index 5efed421ab..39f3977ac5 100644 --- a/src/smolnetd/scheme/mod.rs +++ b/src/smolnetd/scheme/mod.rs @@ -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; diff --git a/src/smolnetd/scheme/socket.rs b/src/smolnetd/scheme/socket.rs index 9b26c1eb07..586e4501a0 100644 --- a/src/smolnetd/scheme/socket.rs +++ b/src/smolnetd/scheme/socket.rs @@ -29,7 +29,7 @@ impl SocketFile { read_timeout: self.read_timeout, write_timeout: self.write_timeout, socket_handle: self.socket_handle, - data + data, } } @@ -40,7 +40,7 @@ impl SocketFile { 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, - write_queue: Vec, -} +type WaitQueue = Vec; + +type WaitQueueMap = BTreeMap; + +pub type DupResult = (SchemeFile, Option<(Socket<'static, 'static>, T::DataT)>); pub trait SchemeSocket where @@ -109,6 +109,8 @@ where fn write_buf(&mut self, &mut SocketFile, buf: &[u8]) -> syscall::Result; fn read_buf(&mut self, &mut SocketFile, buf: &mut [u8]) -> syscall::Result; + fn fpath(&self, &SchemeFile, &mut [u8]) -> syscall::Result; + fn dup( &self, &mut SchemeFile, @@ -116,11 +118,9 @@ where SocketHandle, &str, &mut Self::SchemeDataT, - ) -> syscall::Result>; + ) -> syscall::Result>; } -type WaitQueueMap = BTreeMap; - pub struct SocketScheme 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( + fn wake_up_wait_queue( &mut self, socket_handle: SocketHandle, cur_time: syscall::TimeSpec, - f: F, - ) -> Result<()> - where - F: Fn(&mut WaitQueues) -> &mut Vec, - { - 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::>(); + } + + 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 { 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 { + 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 { 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 { + 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 { + 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 { diff --git a/src/smolnetd/scheme/tcp.rs b/src/smolnetd/scheme/tcp.rs index 1a26e01f39..0889ac7de9 100644 --- a/src/smolnetd/scheme/tcp.rs +++ b/src/smolnetd/scheme/tcp.rs @@ -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, buf: &[u8], ) -> syscall::Result { - 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, buf: &mut [u8], ) -> syscall::Result { - 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> { - trace!("TCP dup {}", path); + ) -> syscall::Result> { 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, buf: &mut [u8]) -> syscall::Result { + 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) } } diff --git a/src/smolnetd/scheme/udp.rs b/src/smolnetd/scheme/udp.rs index 030d53c495..93203e4bca 100644 --- a/src/smolnetd/scheme/udp.rs +++ b/src/smolnetd/scheme/udp.rs @@ -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> { + ) -> syscall::Result> { 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, buf: &mut [u8]) -> syscall::Result { + 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)) + } } }