From de733b4bf86ff85a36e06fbc97ab1838b9723bcd Mon Sep 17 00:00:00 2001 From: Egor Karavaev Date: Mon, 2 Oct 2017 22:11:31 +0300 Subject: [PATCH] Factor out generic socket scheme. --- src/smolnetd/scheme/mod.rs | 5 +- src/smolnetd/scheme/socket.rs | 545 +++++++++++++++++++++++++ src/smolnetd/scheme/udp.rs | 649 ------------------------------ src/smolnetd/scheme/udp_socket.rs | 247 ++++++++++++ 4 files changed, 795 insertions(+), 651 deletions(-) create mode 100644 src/smolnetd/scheme/socket.rs delete mode 100644 src/smolnetd/scheme/udp.rs create mode 100644 src/smolnetd/scheme/udp_socket.rs diff --git a/src/smolnetd/scheme/mod.rs b/src/smolnetd/scheme/mod.rs index 118a862982..04cec36e43 100644 --- a/src/smolnetd/scheme/mod.rs +++ b/src/smolnetd/scheme/mod.rs @@ -14,10 +14,11 @@ use buffer_pool::{Buffer, BufferPool}; use device::NetworkDevice; use error::{Error, Result}; use self::ip::IpScheme; -use self::udp::UdpScheme; +use self::udp_socket::UdpScheme; mod ip; -mod udp; +mod socket; +mod udp_socket; type SocketSet = Rc>>; diff --git a/src/smolnetd/scheme/socket.rs b/src/smolnetd/scheme/socket.rs new file mode 100644 index 0000000000..75ab6a49cc --- /dev/null +++ b/src/smolnetd/scheme/socket.rs @@ -0,0 +1,545 @@ +use smoltcp::socket::{AsSocket, Socket, SocketHandle}; +use std::collections::BTreeMap; +use std::fs::File; +use std::io::{Read, Write}; +use std::str; +use std::marker::PhantomData; +use syscall::SchemeMut; +use syscall::data::TimeSpec; +use syscall; + +use error::{Error, Result}; +use super::SocketSet; +use super::post_fevent; + +pub struct SocketFile { + pub flags: usize, + events: usize, + socket_handle: SocketHandle, + pub data: DataT, + pub read_timeout: Option, + pub write_timeout: Option, +} + +impl SocketFile { + pub fn clone_with_data(&self, data: DataT) -> SocketFile { + SocketFile { + flags: self.flags, + events: self.events, + read_timeout: self.read_timeout, + write_timeout: self.write_timeout, + socket_handle: self.socket_handle, + data + } + } + + pub fn new_with_data(socket_handle: SocketHandle, data: DataT) -> SocketFile { + SocketFile { + flags: 0, + events: 0, + read_timeout: None, + write_timeout: None, + socket_handle, + data + } + } +} + +pub struct SettingFile { + pub fd: usize, + pub socket_handle: SocketHandle, + pub setting: SettingT, +} + +pub enum SchemeFile +where + SocketT: SchemeSocket, +{ + Setting(SettingFile), + Socket(SocketFile), +} + +impl SchemeFile +where + SocketT: SchemeSocket, +{ + fn socket_handle(&self) -> SocketHandle { + match *self { + SchemeFile::Socket(SocketFile { socket_handle, .. }) | + SchemeFile::Setting(SettingFile { socket_handle, .. }) => socket_handle, + } + } +} + +#[derive(Default, Clone)] +struct WaitHandle { + until: Option, + packet: syscall::Packet, +} + +#[derive(Default)] +struct WaitQueues { + read_queue: Vec, + write_queue: Vec, +} + +pub trait SchemeSocket +where + Self: ::std::marker::Sized, +{ + type SchemeDataT; + type DataT; + type SettingT: Copy; + + fn new_scheme_data() -> Self::SchemeDataT; + + fn can_send(&self) -> bool; + fn can_recv(&self) -> bool; + + fn get_setting(&SocketFile, Self::SettingT, &mut [u8]) -> syscall::Result; + fn set_setting(&mut SocketFile, Self::SettingT, &[u8]) -> syscall::Result; + + fn new_socket( + &str, + u32, + &mut Self::SchemeDataT, + ) -> syscall::Result<(Socket<'static, 'static>, Self::DataT)>; + fn close_file(&self, &SchemeFile, &mut Self::SchemeDataT) -> syscall::Result<()>; + + fn write_buf(&mut self, &mut SocketFile, buf: &[u8]) -> syscall::Result; + fn read_buf(&mut self, &mut SocketFile, buf: &mut [u8]) -> syscall::Result; + + fn dup( + &self, + &mut SchemeFile, + usize, + SocketHandle, + &str, + &mut Self::SchemeDataT, + ) -> syscall::Result>; +} + +type WaitQueueMap = BTreeMap; + +pub struct SocketScheme +where + SocketT: SchemeSocket, + Socket<'static, 'static>: AsSocket, +{ + next_fd: usize, + fds: BTreeMap>, + socket_set: SocketSet, + scheme_file: File, + wait_queue_map: WaitQueueMap, + scheme_data: SocketT::SchemeDataT, + _phantom_socket: PhantomData, +} + +impl SocketScheme +where + SocketT: SchemeSocket, + Socket<'static, 'static>: AsSocket, +{ + pub fn new(socket_set: SocketSet, scheme_file: File) -> SocketScheme { + SocketScheme { + next_fd: 1, + fds: BTreeMap::new(), + socket_set, + scheme_data: SocketT::new_scheme_data(), + scheme_file, + wait_queue_map: BTreeMap::new(), + _phantom_socket: PhantomData, + } + } + + pub fn on_scheme_event(&mut self) -> Result> { + loop { + let mut packet = syscall::Packet::default(); + if self.scheme_file.read(&mut packet)? == 0 { + break; + } + let a = packet.a; + self.handle(&mut packet); + if packet.a != (-syscall::EWOULDBLOCK) as usize { + self.scheme_file.write_all(&packet)?; + } else { + packet.a = a; + self.handle_block(packet)?; + } + } + Ok(None) + } + + pub fn notify_sockets(&mut self) -> Result<()> { + // Notify non-blocking sockets + for (&fd, handle) in &self.fds { + if let SchemeFile::Socket(SocketFile { + socket_handle, + events, + .. + }) = *handle + { + let mut socket_set = self.socket_set.borrow_mut(); + let socket: &mut SocketT = socket_set.get_mut(socket_handle).as_socket(); + + if events & syscall::EVENT_READ == syscall::EVENT_READ && socket.can_recv() { + post_fevent(&mut self.scheme_file, fd, syscall::EVENT_READ, 1)?; + } + + if events & syscall::EVENT_WRITE == syscall::EVENT_WRITE && socket.can_send() { + post_fevent(&mut self.scheme_file, fd, syscall::EVENT_WRITE, 1)?; + } + } + } + + // Wake up blocking queue + self.wake_up_queues()?; + + Ok(()) + } + + fn wake_up_queues(&mut self) -> Result<()> { + let mut cur_time = TimeSpec::default(); + syscall::clock_gettime(syscall::CLOCK_MONOTONIC, &mut cur_time) + .map_err(|e| Error::from_syscall_error(e, "Can't get time"))?; + + 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)?; + } + } + Ok(()) + } + + 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) + { + ::std::mem::replace(f(wait_queues), vec![]) + } else { + vec![] + }; + + let mut to_retain = vec![]; + + for wait_handle in input_queue.drain(..) { + let mut packet = wait_handle.packet; + self.handle(&mut packet); + if packet.a == (-syscall::EWOULDBLOCK) as usize { + match wait_handle.until { + Some(until) + if (until.tv_sec >= cur_time.tv_sec + || (until.tv_sec == cur_time.tv_sec + && until.tv_nsec >= cur_time.tv_nsec)) => + { + trace!("Timeouting fd {}", packet.b); + packet.a = (-syscall::ETIMEDOUT) as usize; + self.scheme_file.write_all(&packet)?; + } + _ => { + to_retain.push(wait_handle); + } + } + } else { + trace!("Waking up fd {}", packet.b); + 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); + } + + 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; + self.scheme_file.write_all(&packet)?; + Err(Error::from_syscall_error( + syscall_error, + "Can't handle blocked socket", + )) + } else { + Ok(()) + } + } + + fn try_handle_block(&mut self, packet: &mut syscall::Packet) -> syscall::Result<()> { + let fd = packet.b; + let (socket_handle, read_timeout, write_timeout) = { + let handle = self.fds + .get(&fd) + .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; + + if let SchemeFile::Socket(ref scheme_file) = *handle { + Ok(( + scheme_file.socket_handle, + scheme_file.read_timeout, + scheme_file.write_timeout, + )) + } else { + Err(syscall::Error::new(syscall::EBADF)) + } + }?; + + let mut timeout = match packet.a { + syscall::SYS_READ => Ok(read_timeout), + syscall::SYS_WRITE => Ok(write_timeout), + _ => Err(syscall::Error::new(syscall::EBADF)), + }?; + + if let Some(ref mut timeout) = timeout { + let mut cur_time = TimeSpec::default(); + syscall::clock_gettime(syscall::CLOCK_MONOTONIC, &mut cur_time)?; + *timeout = add_time(timeout, &cur_time) + } + + trace!("Adding {} to wait queie", fd); + let wait_queues = self.wait_queue_map + .entry(socket_handle) + .or_insert_with(|| WaitQueues::default()); + + 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 { + until: timeout, + packet: *packet, + }); + + Ok(()) + } + + fn get_setting( + &mut self, + fd: usize, + setting: SocketT::SettingT, + buf: &mut [u8], + ) -> syscall::Result { + let handle = self.fds + .get_mut(&fd) + .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; + let handle = match *handle { + SchemeFile::Socket(ref mut handle) => handle, + _ => { + return Err(syscall::Error::new(syscall::EBADF)); + } + }; + + SocketT::get_setting(handle, setting, buf) + } + + fn update_setting( + &mut self, + fd: usize, + setting: SocketT::SettingT, + buf: &[u8], + ) -> syscall::Result { + let handle = self.fds + .get_mut(&fd) + .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; + let handle = match *handle { + SchemeFile::Socket(ref mut handle) => handle, + _ => { + return Err(syscall::Error::new(syscall::EBADF)); + } + }; + SocketT::set_setting(handle, setting, buf) + } +} + +impl syscall::SchemeMut for SocketScheme +where + SocketT: SchemeSocket, + Socket<'static, 'static>: AsSocket, +{ + 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)))?; + + let (socket, data) = SocketT::new_socket(path, uid, &mut self.scheme_data)?; + + let socket_handle = self.socket_set.borrow_mut().add(socket); + let id = self.next_fd; + + self.fds.insert( + id, + SchemeFile::Socket(SocketFile { + flags, + events: 0, + socket_handle, + write_timeout: None, + read_timeout: None, + data, + }), + ); + self.next_fd += 1; + Ok(id) + } + + fn close(&mut self, fd: usize) -> syscall::Result { + let socket_handle = { + let handle = self.fds + .get(&fd) + .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; + handle.socket_handle() + }; + let scheme_file = self.fds.remove(&fd); + let mut socket_set = self.socket_set.borrow_mut(); + if let Some(scheme_file) = scheme_file { + let socket: &mut SocketT = socket_set.get_mut(socket_handle).as_socket(); + 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( + |&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() + } else { + false + }; + if remove_wq { + self.wait_queue_map.remove(&socket_handle); + } + socket_set.release(socket_handle); + //TODO: removing sockets in release should make prune unnecessary + socket_set.prune(); + Ok(0) + } + + fn write(&mut self, fd: usize, buf: &[u8]) -> syscall::Result { + let (fd, setting) = { + let handle = self.fds + .get_mut(&fd) + .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; + + match *handle { + SchemeFile::Setting(ref setting_handle) => { + (setting_handle.fd, setting_handle.setting) + } + SchemeFile::Socket(ref mut handle) => { + let mut socket_set = self.socket_set.borrow_mut(); + let socket: &mut SocketT = socket_set.get_mut(handle.socket_handle).as_socket(); + + return ::write_buf(socket, handle, buf); + } + } + }; + self.update_setting(fd, setting, buf) + } + + fn read(&mut self, fd: usize, buf: &mut [u8]) -> syscall::Result { + let (fd, setting) = { + let handle = self.fds + .get_mut(&fd) + .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; + match *handle { + SchemeFile::Setting(ref setting_handle) => { + (setting_handle.fd, setting_handle.setting) + } + SchemeFile::Socket(ref mut handle) => { + let mut socket_set = self.socket_set.borrow_mut(); + let socket: &mut SocketT = socket_set.get_mut(handle.socket_handle).as_socket(); + return ::read_buf(socket, handle, buf); + } + } + }; + self.get_setting(fd, setting, buf) + } + + 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 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) + }?; + + self.socket_set.borrow_mut().retain(handle.socket_handle()); + + let id = self.next_fd; + self.fds.insert(id, handle); + self.next_fd += 1; + + Ok(id) + } + + fn fevent(&mut self, fd: usize, events: usize) -> syscall::Result { + let handle = self.fds + .get_mut(&fd) + .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; + match *handle { + SchemeFile::Setting(_) => Err(syscall::Error::new(syscall::EBADF)), + SchemeFile::Socket(ref mut handle) => { + handle.events = events; + Ok(fd) + } + } + } + + fn fsync(&mut self, fd: usize) -> syscall::Result { + { + let _handle = self.fds + .get_mut(&fd) + .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; + } + Ok(0) + // TODO Implement fsyncing + // self.0.network_fsync() + } +} + +fn add_time(a: &TimeSpec, b: &TimeSpec) -> TimeSpec { + let mut secs = a.tv_sec + b.tv_sec; + let mut nsecs = a.tv_nsec + b.tv_nsec; + + secs += i64::from(nsecs) / 1_000_000_000; + nsecs %= 1_000_000_000; + + TimeSpec { + tv_sec: secs, + tv_nsec: nsecs, + } +} diff --git a/src/smolnetd/scheme/udp.rs b/src/smolnetd/scheme/udp.rs deleted file mode 100644 index 764af80d5e..0000000000 --- a/src/smolnetd/scheme/udp.rs +++ /dev/null @@ -1,649 +0,0 @@ -use smoltcp::socket::{AsSocket, SocketHandle, UdpPacketBuffer, UdpSocket, UdpSocketBuffer}; -use smoltcp::wire::{IpAddress, IpEndpoint, Ipv4Address}; -use std::collections::BTreeMap; -use std::fs::File; -use std::io::{Read, Write}; -use std::str::FromStr; -use std::str; -use std::mem; -use std::ops::Deref; -use std::ops::DerefMut; -use syscall::SchemeMut; -use syscall::data::TimeSpec; -use syscall; - -use device::NetworkDevice; -use error::{Error, Result}; -use super::{Smolnetd, SocketSet}; -use port_set::PortSet; -use super::post_fevent; - -#[derive(Copy, Clone, Eq, PartialEq, Debug)] -enum Setting { - Ttl, - ReadTimeout, - WriteTimeout, -} - -struct UdpHandle { - flags: usize, - events: usize, - remote_endpoint: IpEndpoint, - read_timeout: Option, - write_timeout: Option, - socket_handle: SocketHandle, -} - -struct SettingHandle { - fd: usize, - socket_handle: SocketHandle, - setting: Setting, -} - -enum FdHandle { - Setting(SettingHandle), - Socket(UdpHandle), -} - -impl FdHandle { - fn socket_handle(&self) -> SocketHandle { - match *self { - FdHandle::Socket(UdpHandle { socket_handle, .. }) | - FdHandle::Setting(SettingHandle { socket_handle, .. }) => socket_handle, - } - } -} - -#[derive(Default, Clone)] -struct WaitHandle { - until: Option, - packet: syscall::Packet, -} - -#[derive(Default)] -struct WaitQueues { - read_queue: Vec, - write_queue: Vec, -} - -type WaitQueueMap = BTreeMap; - -pub struct UdpScheme { - next_udp_fd: usize, - udp_fds: BTreeMap, - socket_set: SocketSet, - port_set: PortSet, - udp_file: File, - wait_queue_map: WaitQueueMap, -} - -impl UdpScheme { - pub fn new(socket_set: SocketSet, udp_file: File) -> UdpScheme { - UdpScheme { - next_udp_fd: 1, - udp_fds: BTreeMap::new(), - socket_set, - // 49152..65535 is the suggested range for dynamic private ports - port_set: PortSet::new(49_152u16, 65_535u16).expect("Wrong UDP port numbers"), - udp_file, - wait_queue_map: BTreeMap::new(), - } - } - - pub fn on_scheme_event(&mut self) -> Result> { - loop { - let mut packet = syscall::Packet::default(); - if self.udp_file.read(&mut packet)? == 0 { - break; - } - let a = packet.a; - self.handle(&mut packet); - if packet.a != (-syscall::EWOULDBLOCK) as usize { - self.udp_file.write_all(&packet)?; - } else { - packet.a = a; - self.handle_block(packet)?; - } - } - Ok(None) - } - - pub fn notify_sockets(&mut self) -> Result<()> { - // Notify non-blocking sockets - for (&fd, handle) in &self.udp_fds { - if let FdHandle::Socket(UdpHandle { - socket_handle, - events, - .. - }) = *handle - { - let mut socket_set = self.socket_set.borrow_mut(); - let socket: &mut UdpSocket = socket_set.get_mut(socket_handle).as_socket(); - - if events & syscall::EVENT_READ == syscall::EVENT_READ && socket.can_recv() { - post_fevent(&mut self.udp_file, fd, syscall::EVENT_READ, 1)?; - } - - if events & syscall::EVENT_WRITE == syscall::EVENT_WRITE && socket.can_send() { - post_fevent(&mut self.udp_file, fd, syscall::EVENT_WRITE, 1)?; - } - } - } - - // Wake up blocking queue - self.wake_up_queues()?; - - Ok(()) - } - - fn wake_up_queues(&mut self) -> Result<()> { - let mut cur_time = TimeSpec::default(); - syscall::clock_gettime(syscall::CLOCK_MONOTONIC, &mut cur_time) - .map_err(|e| Error::from_syscall_error(e, "Can't get time"))?; - - 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 UdpSocket = 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)?; - } - } - Ok(()) - } - - 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) - { - ::std::mem::replace(f(wait_queues), vec![]) - } else { - vec![] - }; - - let mut to_retain = vec![]; - - for wait_handle in input_queue.drain(..) { - let mut packet = wait_handle.packet; - self.handle(&mut packet); - if packet.a == (-syscall::EWOULDBLOCK) as usize { - match wait_handle.until { - Some(until) - if (until.tv_sec >= cur_time.tv_sec - || (until.tv_sec == cur_time.tv_sec - && until.tv_nsec >= cur_time.tv_nsec)) => - { - trace!("Timeouting fd {}", packet.b); - packet.a = (-syscall::ETIMEDOUT) as usize; - self.udp_file.write_all(&packet)?; - } - _ => { - to_retain.push(wait_handle); - } - } - } else { - trace!("Waking up fd {}", packet.b); - self.udp_file.write_all(&packet)?; - } - } - - if let Some(wait_queues) = self.wait_queue_map.get_mut(&socket_handle) { - f(wait_queues).extend(to_retain); - } - - 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; - self.udp_file.write_all(&packet)?; - Err(Error::from_syscall_error( - syscall_error, - "Can't handle blocked socket", - )) - } else { - Ok(()) - } - } - - fn try_handle_block(&mut self, packet: &mut syscall::Packet) -> syscall::Result<()> { - let fd = packet.b; - let (socket_handle, read_timeout, write_timeout) = { - let handle = self.udp_fds - .get(&fd) - .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; - - if let FdHandle::Socket(ref udp_handle) = *handle { - Ok(( - udp_handle.socket_handle, - udp_handle.read_timeout, - udp_handle.write_timeout, - )) - } else { - Err(syscall::Error::new(syscall::EBADF)) - } - }?; - - let mut timeout = match packet.a { - syscall::SYS_READ => Ok(read_timeout), - syscall::SYS_WRITE => Ok(write_timeout), - _ => Err(syscall::Error::new(syscall::EBADF)), - }?; - - if let Some(ref mut timeout) = timeout { - let mut cur_time = TimeSpec::default(); - syscall::clock_gettime(syscall::CLOCK_MONOTONIC, &mut cur_time)?; - *timeout = add_time(timeout, &cur_time) - } - - trace!("Adding {} to wait queie", fd); - let wait_queues = self.wait_queue_map - .entry(socket_handle) - .or_insert_with(|| WaitQueues::default()); - - 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 { - until: timeout, - packet: *packet, - }); - - Ok(()) - } - - fn get_setting( - &mut self, - fd: usize, - setting: Setting, - buf: &mut [u8], - ) -> syscall::Result { - let handle = self.udp_fds - .get_mut(&fd) - .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; - let handle = match *handle { - FdHandle::Socket(ref mut handle) => handle, - _ => { - return Err(syscall::Error::new(syscall::EBADF)); - } - }; - let timespec = match (setting, handle.read_timeout, handle.write_timeout) { - (Setting::ReadTimeout, Some(read_timeout), _) => read_timeout, - (Setting::WriteTimeout, _, Some(write_timeout)) => write_timeout, - _ => { - return Ok(0); - } - }; - - trace!("Getting {:?} from {}", setting, fd); - - if buf.len() < mem::size_of::() { - Ok(0) - } else { - let count = timespec.deref().read(buf).map_err(|err| { - syscall::Error::new(err.raw_os_error().unwrap_or(syscall::EIO)) - })?; - Ok(count) - } - } - - fn update_setting( - &mut self, - fd: usize, - setting: Setting, - buf: &[u8], - ) -> syscall::Result { - let handle = self.udp_fds - .get_mut(&fd) - .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; - let handle = match *handle { - FdHandle::Socket(ref mut handle) => handle, - _ => { - return Err(syscall::Error::new(syscall::EBADF)); - } - }; - trace!("Setting {:?} to {}", setting, fd); - match setting { - Setting::ReadTimeout | Setting::WriteTimeout => { - let (timeout, count) = { - if buf.len() < mem::size_of::() { - (None, 0) - } else { - let mut timespec = TimeSpec::default(); - let count = timespec.deref_mut().write(buf).map_err(|err| { - syscall::Error::new(err.raw_os_error().unwrap_or(syscall::EIO)) - })?; - (Some(timespec), count) - } - }; - match setting { - Setting::ReadTimeout => { - handle.read_timeout = timeout; - } - Setting::WriteTimeout => { - handle.write_timeout = timeout; - } - _ => {} - }; - return Ok(count); - } - Setting::Ttl => {} - } - Ok(0) - } -} - -impl syscall::SchemeMut for UdpScheme { - 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!("Udp open {} ", path); - - let mut parts = path.split('/'); - let remote_endpoint = parse_endpoint(parts.next().unwrap_or("")); - let mut local_endpoint = parse_endpoint(parts.next().unwrap_or("")); - - if local_endpoint.port <= 1024 && uid != 0 { - return Err(syscall::Error::new(syscall::EACCES)); - } - - let mut rx_packets = Vec::with_capacity(Smolnetd::SOCKET_BUFFER_SIZE); - let mut tx_packets = Vec::with_capacity(Smolnetd::SOCKET_BUFFER_SIZE); - for _ in 0..Smolnetd::SOCKET_BUFFER_SIZE { - rx_packets.push(UdpPacketBuffer::new(vec![0; NetworkDevice::MTU])); - tx_packets.push(UdpPacketBuffer::new(vec![0; NetworkDevice::MTU])); - } - let rx_buffer = UdpSocketBuffer::new(rx_packets); - let tx_buffer = UdpSocketBuffer::new(tx_packets); - let mut udp_socket = UdpSocket::new(rx_buffer, tx_buffer); - - if local_endpoint.port == 0 { - local_endpoint.port = self.port_set - .get_port() - .ok_or_else(|| syscall::Error::new(syscall::EINVAL))?; - } else if !self.port_set.claim_port(local_endpoint.port) { - return Err(syscall::Error::new(syscall::EADDRINUSE)); - } - - { - let udp_socket: &mut UdpSocket = udp_socket.as_socket(); - udp_socket - .bind(local_endpoint) - .expect("Can't bind udp socket to local endpoint"); - } - - let socket_handle = self.socket_set.borrow_mut().add(udp_socket); - let id = self.next_udp_fd; - - self.udp_fds.insert( - id, - FdHandle::Socket(UdpHandle { - flags, - events: 0, - socket_handle, - write_timeout: None, - read_timeout: None, - remote_endpoint: remote_endpoint, - }), - ); - self.next_udp_fd += 1; - trace!("Udp open fd {} ", id); - Ok(id) - } - - fn close(&mut self, fd: usize) -> syscall::Result { - trace!("Upd close {}", fd); - let socket_handle = { - let handle = self.udp_fds - .get(&fd) - .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; - handle.socket_handle() - }; - self.udp_fds.remove(&fd); - let mut socket_set = self.socket_set.borrow_mut(); - let endpoint = { - let socket: &mut UdpSocket = socket_set.get_mut(socket_handle).as_socket(); - socket.endpoint() - }; - let remove = if let Some(ref mut wait_queues) = self.wait_queue_map.get_mut(&socket_handle) - { - wait_queues.read_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() - } else { - false - }; - if remove { - self.wait_queue_map.remove(&socket_handle); - } - socket_set.release(socket_handle); - //TODO: removing sockets in release should make prune unnecessary - socket_set.prune(); - if endpoint.port != 0 { - self.port_set.release_port(endpoint.port); - } - Ok(0) - } - - fn write(&mut self, fd: usize, buf: &[u8]) -> syscall::Result { - trace!("Upd write {} len {}", fd, buf.len()); - - let (fd, setting) = { - let handle = self.udp_fds - .get(&fd) - .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; - - match *handle { - FdHandle::Setting(ref setting_handle) => { - (setting_handle.fd, setting_handle.setting) - } - FdHandle::Socket(ref handle) => { - if !handle.remote_endpoint.is_specified() { - return Err(syscall::Error::new(syscall::EADDRNOTAVAIL)); - } - let mut socket_set = self.socket_set.borrow_mut(); - let socket: &mut UdpSocket = - socket_set.get_mut(handle.socket_handle).as_socket(); - return if socket.can_send() { - socket - .send_slice(buf, handle.remote_endpoint) - .expect("Can't send slice"); - Ok(buf.len()) - } else if handle.flags & syscall::O_NONBLOCK == syscall::O_NONBLOCK { - Ok(0) - } else { - Err(syscall::Error::new(syscall::EWOULDBLOCK)) - }; - } - } - }; - self.update_setting(fd, setting, buf) - } - - fn read(&mut self, fd: usize, buf: &mut [u8]) -> syscall::Result { - trace!("udp read {} {}", fd, buf.len()); - let (fd, setting) = { - let handle = self.udp_fds.get(&fd).ok_or_else(|| { - trace!("UDP read EBADF {}", fd); - syscall::Error::new(syscall::EBADF) - })?; - match *handle { - FdHandle::Setting(ref setting_handle) => { - (setting_handle.fd, setting_handle.setting) - } - FdHandle::Socket(ref handle) => { - let mut socket_set = self.socket_set.borrow_mut(); - let socket: &mut UdpSocket = - socket_set.get_mut(handle.socket_handle).as_socket(); - return if socket.can_recv() { - let (length, _) = socket.recv_slice(buf).expect("Can't receive slice"); - trace!("Upd read {}", fd); - Ok(length) - } else if handle.flags & syscall::O_NONBLOCK == syscall::O_NONBLOCK { - Ok(0) - } else { - Err(syscall::Error::new(syscall::EWOULDBLOCK)) - }; - } - } - }; - self.get_setting(fd, setting, buf) - } - - fn dup(&mut self, fd: usize, buf: &[u8]) -> syscall::Result { - let path = str::from_utf8(buf).or_else(|_| Err(syscall::Error::new(syscall::EINVAL)))?; - trace!("udp dup {} {}", fd, path); - - let handle = { - let handle = self.udp_fds - .get_mut(&fd) - .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; - - let socket_handle = handle.socket_handle(); - - match path { - "ttl" => FdHandle::Setting(SettingHandle { - socket_handle, - fd, - setting: Setting::Ttl, - }), - "read_timeout" => FdHandle::Setting(SettingHandle { - socket_handle, - fd, - setting: Setting::ReadTimeout, - }), - "write_timeout" => FdHandle::Setting(SettingHandle { - socket_handle, - fd, - setting: Setting::WriteTimeout, - }), - _ => { - let remote_endpoint = parse_endpoint(path); - if let FdHandle::Socket(ref udp_handle) = *handle { - FdHandle::Socket(UdpHandle { - flags: udp_handle.flags, - events: udp_handle.events, - read_timeout: udp_handle.read_timeout, - write_timeout: udp_handle.write_timeout, - remote_endpoint: if remote_endpoint.is_specified() { - remote_endpoint - } else { - udp_handle.remote_endpoint - }, - socket_handle, - }) - } else { - FdHandle::Socket(UdpHandle { - flags: 0, - events: 0, - read_timeout: None, - write_timeout: None, - remote_endpoint: remote_endpoint, - socket_handle, - }) - } - } - } - }; - - self.socket_set.borrow_mut().retain(handle.socket_handle()); - let port = { - let mut socket_set = self.socket_set.borrow_mut(); - let socket: &mut UdpSocket = socket_set.get_mut(handle.socket_handle()).as_socket(); - socket.endpoint().port - }; - self.port_set.acquire_port(port); - - let id = self.next_udp_fd; - self.udp_fds.insert(id, handle); - self.next_udp_fd += 1; - - Ok(id) - } - - fn fevent(&mut self, fd: usize, events: usize) -> syscall::Result { - trace!("udp fevent {}", fd); - let handle = self.udp_fds - .get_mut(&fd) - .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; - match *handle { - FdHandle::Setting(_) => Err(syscall::Error::new(syscall::EBADF)), - FdHandle::Socket(ref mut handle) => { - handle.events = events; - Ok(fd) - } - } - } - - fn fsync(&mut self, fd: usize) -> syscall::Result { - trace!("udp fsync {}", fd); - { - let _handle = self.udp_fds - .get_mut(&fd) - .ok_or_else(|| syscall::Error::new(syscall::EBADF))?; - } - Ok(0) - // TODO Implement fsyncing - // self.0.network_fsync() - } -} - -fn parse_endpoint(socket: &str) -> IpEndpoint { - let mut socket_parts = socket.split(':'); - let host = IpAddress::Ipv4( - Ipv4Address::from_str(socket_parts.next().unwrap_or("")) - .unwrap_or_else(|_| Ipv4Address::new(0, 0, 0, 0)), - ); - - let port = socket_parts - .next() - .unwrap_or("") - .parse::() - .unwrap_or(0); - IpEndpoint::new(host, port) -} - -fn add_time(a: &TimeSpec, b: &TimeSpec) -> TimeSpec { - let mut secs = a.tv_sec + b.tv_sec; - let mut nsecs = a.tv_nsec + b.tv_nsec; - - secs += i64::from(nsecs) / 1_000_000_000; - nsecs %= 1_000_000_000; - - TimeSpec { - tv_sec: secs, - tv_nsec: nsecs, - } -} diff --git a/src/smolnetd/scheme/udp_socket.rs b/src/smolnetd/scheme/udp_socket.rs new file mode 100644 index 0000000000..e06825a1e3 --- /dev/null +++ b/src/smolnetd/scheme/udp_socket.rs @@ -0,0 +1,247 @@ +use super::socket::*; + +use smoltcp::socket::{AsSocket, Socket, SocketHandle, UdpPacketBuffer, UdpSocket, UdpSocketBuffer}; +use smoltcp::wire::{IpAddress, IpEndpoint, Ipv4Address}; +use std::io::{Read, Write}; +use std::str::FromStr; +use std::str; +use std::mem; +use std::ops::Deref; +use std::ops::DerefMut; +use syscall::data::TimeSpec; +use syscall; + +use device::NetworkDevice; +use super::Smolnetd; +use port_set::PortSet; + +pub type UdpScheme = SocketScheme>; + +#[derive(Copy, Clone, Eq, PartialEq, Debug)] +pub enum Setting { + Ttl, + ReadTimeout, + WriteTimeout, +} + +impl<'a, 'b> SchemeSocket for UdpSocket<'a, 'b> { + type SchemeDataT = PortSet; + type DataT = IpEndpoint; + type SettingT = Setting; + + fn new_scheme_data() -> Self::SchemeDataT { + PortSet::new(49_152u16, 65_535u16).expect("Wrong UDP port numbers") + } + + fn can_send(&self) -> bool { + self.can_send() + } + + fn can_recv(&self) -> bool { + self.can_recv() + } + + fn get_setting( + file: &SocketFile, + setting: Self::SettingT, + buf: &mut [u8], + ) -> syscall::Result { + let timespec = match (setting, file.read_timeout, file.write_timeout) { + (Setting::ReadTimeout, Some(read_timeout), _) => read_timeout, + (Setting::WriteTimeout, _, Some(write_timeout)) => write_timeout, + _ => { + return Ok(0); + } + }; + + if buf.len() < mem::size_of::() { + Ok(0) + } else { + let count = timespec.deref().read(buf).map_err(|err| { + syscall::Error::new(err.raw_os_error().unwrap_or(syscall::EIO)) + })?; + Ok(count) + } + } + + fn set_setting( + file: &mut SocketFile, + setting: Self::SettingT, + buf: &[u8], + ) -> syscall::Result { + match setting { + Setting::ReadTimeout | Setting::WriteTimeout => { + let (timeout, count) = { + if buf.len() < mem::size_of::() { + (None, 0) + } else { + let mut timespec = TimeSpec::default(); + let count = timespec.deref_mut().write(buf).map_err(|err| { + syscall::Error::new(err.raw_os_error().unwrap_or(syscall::EIO)) + })?; + (Some(timespec), count) + } + }; + match setting { + Setting::ReadTimeout => { + file.read_timeout = timeout; + } + Setting::WriteTimeout => { + file.write_timeout = timeout; + } + _ => {} + }; + return Ok(count); + } + Setting::Ttl => {} + } + Ok(0) + } + + fn new_socket( + path: &str, + uid: u32, + port_set: &mut Self::SchemeDataT, + ) -> syscall::Result<(Socket<'static, 'static>, Self::DataT)> { + let mut parts = path.split('/'); + let remote_endpoint = parse_endpoint(parts.next().unwrap_or("")); + let mut local_endpoint = parse_endpoint(parts.next().unwrap_or("")); + + if local_endpoint.port <= 1024 && uid != 0 { + return Err(syscall::Error::new(syscall::EACCES)); + } + + let mut rx_packets = Vec::with_capacity(Smolnetd::SOCKET_BUFFER_SIZE); + let mut tx_packets = Vec::with_capacity(Smolnetd::SOCKET_BUFFER_SIZE); + for _ in 0..Smolnetd::SOCKET_BUFFER_SIZE { + rx_packets.push(UdpPacketBuffer::new(vec![0; NetworkDevice::MTU])); + tx_packets.push(UdpPacketBuffer::new(vec![0; NetworkDevice::MTU])); + } + let rx_buffer = UdpSocketBuffer::new(rx_packets); + let tx_buffer = UdpSocketBuffer::new(tx_packets); + let mut udp_socket = UdpSocket::new(rx_buffer, tx_buffer); + + if local_endpoint.port == 0 { + local_endpoint.port = port_set + .get_port() + .ok_or_else(|| syscall::Error::new(syscall::EINVAL))?; + } else if !port_set.claim_port(local_endpoint.port) { + return Err(syscall::Error::new(syscall::EADDRINUSE)); + } + + { + let udp_socket: &mut UdpSocket = udp_socket.as_socket(); + udp_socket + .bind(local_endpoint) + .expect("Can't bind udp socket to local endpoint"); + } + + Ok((udp_socket, remote_endpoint)) + } + + fn close_file( + &self, + file: &SchemeFile, + port_set: &mut Self::SchemeDataT, + ) -> syscall::Result<()> { + if let &SchemeFile::Socket(_) = file { + port_set.release_port(self.endpoint().port); + } + Ok(()) + } + + fn write_buf( + &mut self, + file: &mut SocketFile, + buf: &[u8], + ) -> syscall::Result { + if !file.data.is_specified() { + return Err(syscall::Error::new(syscall::EADDRNOTAVAIL)); + } + return if self.can_send() { + self.send_slice(buf, file.data).expect("Can't send slice"); + Ok(buf.len()) + } else if file.flags & syscall::O_NONBLOCK == syscall::O_NONBLOCK { + Ok(0) + } else { + Err(syscall::Error::new(syscall::EWOULDBLOCK)) + }; + } + + fn read_buf( + &mut self, + file: &mut SocketFile, + buf: &mut [u8], + ) -> syscall::Result { + return 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 { + Ok(0) + } else { + Err(syscall::Error::new(syscall::EWOULDBLOCK)) + }; + } + + fn dup( + &self, + file: &mut SchemeFile, + fd: usize, + socket_handle: SocketHandle, + path: &str, + port_set: &mut Self::SchemeDataT, + ) -> syscall::Result> { + let handle = match path { + "ttl" => SchemeFile::Setting(SettingFile { + socket_handle, + fd, + setting: Setting::Ttl, + }), + "read_timeout" => SchemeFile::Setting(SettingFile { + socket_handle, + fd, + setting: Setting::ReadTimeout, + }), + "write_timeout" => SchemeFile::Setting(SettingFile { + socket_handle, + fd, + setting: Setting::WriteTimeout, + }), + _ => { + let remote_endpoint = parse_endpoint(path); + if let SchemeFile::Socket(ref udp_handle) = *file { + SchemeFile::Socket(udp_handle.clone_with_data( + if remote_endpoint.is_specified() { + remote_endpoint + } else { + udp_handle.data + }, + )) + } else { + SchemeFile::Socket(SocketFile::new_with_data(socket_handle, remote_endpoint)) + } + } + }; + + if let SchemeFile::Socket(_) = handle { + port_set.acquire_port(self.endpoint().port); + } + + Ok(handle) + } +} + +fn parse_endpoint(socket: &str) -> IpEndpoint { + let mut socket_parts = socket.split(':'); + let host = IpAddress::Ipv4( + Ipv4Address::from_str(socket_parts.next().unwrap_or("")) + .unwrap_or_else(|_| Ipv4Address::new(0, 0, 0, 0)), + ); + + let port = socket_parts + .next() + .unwrap_or("") + .parse::() + .unwrap_or(0); + IpEndpoint::new(host, port) +}