Factor out generic socket scheme.
This commit is contained in:
Egor Karavaev
@@ -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<RefCell<smoltcp::socket::SocketSet<'static, 'static, 'static>>>;
|
||||
|
||||
|
||||
@@ -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<DataT> {
|
||||
pub flags: usize,
|
||||
events: usize,
|
||||
socket_handle: SocketHandle,
|
||||
pub data: DataT,
|
||||
pub read_timeout: Option<TimeSpec>,
|
||||
pub write_timeout: Option<TimeSpec>,
|
||||
}
|
||||
|
||||
impl<DataT> SocketFile<DataT> {
|
||||
pub fn clone_with_data(&self, data: DataT) -> SocketFile<DataT> {
|
||||
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<DataT> {
|
||||
SocketFile {
|
||||
flags: 0,
|
||||
events: 0,
|
||||
read_timeout: None,
|
||||
write_timeout: None,
|
||||
socket_handle,
|
||||
data
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct SettingFile<SettingT> {
|
||||
pub fd: usize,
|
||||
pub socket_handle: SocketHandle,
|
||||
pub setting: SettingT,
|
||||
}
|
||||
|
||||
pub enum SchemeFile<SocketT>
|
||||
where
|
||||
SocketT: SchemeSocket,
|
||||
{
|
||||
Setting(SettingFile<SocketT::SettingT>),
|
||||
Socket(SocketFile<SocketT::DataT>),
|
||||
}
|
||||
|
||||
impl<SocketT> SchemeFile<SocketT>
|
||||
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<TimeSpec>,
|
||||
packet: syscall::Packet,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct WaitQueues {
|
||||
read_queue: Vec<WaitHandle>,
|
||||
write_queue: Vec<WaitHandle>,
|
||||
}
|
||||
|
||||
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::DataT>, Self::SettingT, &mut [u8]) -> syscall::Result<usize>;
|
||||
fn set_setting(&mut SocketFile<Self::DataT>, Self::SettingT, &[u8]) -> syscall::Result<usize>;
|
||||
|
||||
fn new_socket(
|
||||
&str,
|
||||
u32,
|
||||
&mut Self::SchemeDataT,
|
||||
) -> syscall::Result<(Socket<'static, 'static>, Self::DataT)>;
|
||||
fn close_file(&self, &SchemeFile<Self>, &mut Self::SchemeDataT) -> syscall::Result<()>;
|
||||
|
||||
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 dup(
|
||||
&self,
|
||||
&mut SchemeFile<Self>,
|
||||
usize,
|
||||
SocketHandle,
|
||||
&str,
|
||||
&mut Self::SchemeDataT,
|
||||
) -> syscall::Result<SchemeFile<Self>>;
|
||||
}
|
||||
|
||||
type WaitQueueMap = BTreeMap<SocketHandle, WaitQueues>;
|
||||
|
||||
pub struct SocketScheme<SocketT>
|
||||
where
|
||||
SocketT: SchemeSocket,
|
||||
Socket<'static, 'static>: AsSocket<SocketT>,
|
||||
{
|
||||
next_fd: usize,
|
||||
fds: BTreeMap<usize, SchemeFile<SocketT>>,
|
||||
socket_set: SocketSet,
|
||||
scheme_file: File,
|
||||
wait_queue_map: WaitQueueMap,
|
||||
scheme_data: SocketT::SchemeDataT,
|
||||
_phantom_socket: PhantomData<SocketT>,
|
||||
}
|
||||
|
||||
impl<SocketT> SocketScheme<SocketT>
|
||||
where
|
||||
SocketT: SchemeSocket,
|
||||
Socket<'static, 'static>: AsSocket<SocketT>,
|
||||
{
|
||||
pub fn new(socket_set: SocketSet, scheme_file: File) -> SocketScheme<SocketT> {
|
||||
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<Option<()>> {
|
||||
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<F>(
|
||||
&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)
|
||||
{
|
||||
::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<usize> {
|
||||
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<usize> {
|
||||
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<SocketT> syscall::SchemeMut for SocketScheme<SocketT>
|
||||
where
|
||||
SocketT: SchemeSocket,
|
||||
Socket<'static, 'static>: AsSocket<SocketT>,
|
||||
{
|
||||
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)))?;
|
||||
|
||||
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<usize> {
|
||||
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<usize> {
|
||||
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 <SocketT as SchemeSocket>::write_buf(socket, handle, buf);
|
||||
}
|
||||
}
|
||||
};
|
||||
self.update_setting(fd, setting, buf)
|
||||
}
|
||||
|
||||
fn read(&mut self, fd: usize, buf: &mut [u8]) -> syscall::Result<usize> {
|
||||
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 <SocketT as SchemeSocket>::read_buf(socket, handle, buf);
|
||||
}
|
||||
}
|
||||
};
|
||||
self.get_setting(fd, setting, buf)
|
||||
}
|
||||
|
||||
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 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<usize> {
|
||||
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<usize> {
|
||||
{
|
||||
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,
|
||||
}
|
||||
}
|
||||
@@ -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<TimeSpec>,
|
||||
write_timeout: Option<TimeSpec>,
|
||||
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<TimeSpec>,
|
||||
packet: syscall::Packet,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct WaitQueues {
|
||||
read_queue: Vec<WaitHandle>,
|
||||
write_queue: Vec<WaitHandle>,
|
||||
}
|
||||
|
||||
type WaitQueueMap = BTreeMap<SocketHandle, WaitQueues>;
|
||||
|
||||
pub struct UdpScheme {
|
||||
next_udp_fd: usize,
|
||||
udp_fds: BTreeMap<usize, FdHandle>,
|
||||
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<Option<()>> {
|
||||
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<F>(
|
||||
&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)
|
||||
{
|
||||
::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<usize> {
|
||||
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::<TimeSpec>() {
|
||||
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<usize> {
|
||||
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::<TimeSpec>() {
|
||||
(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<usize> {
|
||||
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<usize> {
|
||||
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<usize> {
|
||||
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<usize> {
|
||||
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<usize> {
|
||||
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<usize> {
|
||||
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<usize> {
|
||||
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::<u16>()
|
||||
.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,
|
||||
}
|
||||
}
|
||||
@@ -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<UdpSocket<'static, 'static>>;
|
||||
|
||||
#[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<Self::DataT>,
|
||||
setting: Self::SettingT,
|
||||
buf: &mut [u8],
|
||||
) -> syscall::Result<usize> {
|
||||
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::<TimeSpec>() {
|
||||
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<Self::DataT>,
|
||||
setting: Self::SettingT,
|
||||
buf: &[u8],
|
||||
) -> syscall::Result<usize> {
|
||||
match setting {
|
||||
Setting::ReadTimeout | Setting::WriteTimeout => {
|
||||
let (timeout, count) = {
|
||||
if buf.len() < mem::size_of::<TimeSpec>() {
|
||||
(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<Self>,
|
||||
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<Self::DataT>,
|
||||
buf: &[u8],
|
||||
) -> syscall::Result<usize> {
|
||||
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<Self::DataT>,
|
||||
buf: &mut [u8],
|
||||
) -> syscall::Result<usize> {
|
||||
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<Self>,
|
||||
fd: usize,
|
||||
socket_handle: SocketHandle,
|
||||
path: &str,
|
||||
port_set: &mut Self::SchemeDataT,
|
||||
) -> syscall::Result<SchemeFile<Self>> {
|
||||
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::<u16>()
|
||||
.unwrap_or(0);
|
||||
IpEndpoint::new(host, port)
|
||||
}
|
||||
Reference in New Issue
Block a user