mirror of
https://github.com/asterinas/asterinas.git
synced 2025-06-23 01:13:23 +00:00
Refactor unix stream socket implementation
This commit is contained in:
Jianfeng Jiang
committed by
Tate, Hongliang Tian
Tate, Hongliang Tian
parent
93429ae2c9
commit
3f15bcaf5d
@ -1,55 +1,50 @@
|
||||
use crate::{
|
||||
net::socket::{unix::addr::UnixSocketAddr, SockShutdownCmd},
|
||||
prelude::*,
|
||||
};
|
||||
|
||||
use super::endpoint::Endpoint;
|
||||
use crate::fs::utils::{IoEvents, Poller};
|
||||
use crate::net::socket::{unix::addr::UnixSocketAddrBound, SockShutdownCmd};
|
||||
use crate::prelude::*;
|
||||
|
||||
pub struct Connected {
|
||||
pub(super) struct Connected {
|
||||
local_endpoint: Arc<Endpoint>,
|
||||
// The peer addr is None if peer is unnamed.
|
||||
// FIXME: can a socket be bound after the socket is connected?
|
||||
peer_addr: Option<UnixSocketAddr>,
|
||||
}
|
||||
|
||||
impl Connected {
|
||||
pub fn new(local_endpoint: Arc<Endpoint>) -> Self {
|
||||
let peer_addr = local_endpoint.peer_addr();
|
||||
Connected {
|
||||
local_endpoint,
|
||||
peer_addr,
|
||||
}
|
||||
pub(super) fn new(local_endpoint: Arc<Endpoint>) -> Self {
|
||||
Connected { local_endpoint }
|
||||
}
|
||||
|
||||
pub fn addr(&self) -> Option<UnixSocketAddr> {
|
||||
pub(super) fn addr(&self) -> Option<UnixSocketAddrBound> {
|
||||
self.local_endpoint.addr()
|
||||
}
|
||||
|
||||
pub fn peer_addr(&self) -> Option<&UnixSocketAddr> {
|
||||
self.peer_addr.as_ref()
|
||||
pub(super) fn peer_addr(&self) -> Option<UnixSocketAddrBound> {
|
||||
self.local_endpoint.peer_addr()
|
||||
}
|
||||
|
||||
pub fn is_bound(&self) -> bool {
|
||||
pub(super) fn is_bound(&self) -> bool {
|
||||
self.addr().is_some()
|
||||
}
|
||||
|
||||
pub fn write(&self, buf: &[u8]) -> Result<usize> {
|
||||
pub(super) fn write(&self, buf: &[u8]) -> Result<usize> {
|
||||
self.local_endpoint.write(buf)
|
||||
}
|
||||
|
||||
pub fn read(&self, buf: &mut [u8]) -> Result<usize> {
|
||||
pub(super) fn read(&self, buf: &mut [u8]) -> Result<usize> {
|
||||
self.local_endpoint.read(buf)
|
||||
}
|
||||
|
||||
pub fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
|
||||
pub(super) fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
|
||||
self.local_endpoint.shutdown(cmd)
|
||||
}
|
||||
|
||||
pub fn is_nonblocking(&self) -> bool {
|
||||
pub(super) fn is_nonblocking(&self) -> bool {
|
||||
self.local_endpoint.is_nonblocking()
|
||||
}
|
||||
|
||||
pub fn set_nonblocking(&self, is_nonblocking: bool) {
|
||||
pub(super) fn set_nonblocking(&self, is_nonblocking: bool) {
|
||||
self.local_endpoint.set_nonblocking(is_nonblocking).unwrap();
|
||||
}
|
||||
|
||||
pub(super) fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
|
||||
self.local_endpoint.poll(mask, poller)
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,20 +1,20 @@
|
||||
use crate::{
|
||||
fs::utils::{Channel, Consumer, IoEvents, Poller, Producer, StatusFlags},
|
||||
net::socket::{unix::addr::UnixSocketAddr, SockShutdownCmd},
|
||||
net::socket::{unix::addr::UnixSocketAddrBound, SockShutdownCmd},
|
||||
prelude::*,
|
||||
};
|
||||
|
||||
pub struct Endpoint(Inner);
|
||||
pub(super) struct Endpoint(Inner);
|
||||
|
||||
struct Inner {
|
||||
addr: RwLock<Option<UnixSocketAddr>>,
|
||||
addr: RwLock<Option<UnixSocketAddrBound>>,
|
||||
reader: Consumer<u8>,
|
||||
writer: Producer<u8>,
|
||||
peer: Weak<Endpoint>,
|
||||
}
|
||||
|
||||
impl Endpoint {
|
||||
pub fn end_pair(is_nonblocking: bool) -> Result<(Arc<Endpoint>, Arc<Endpoint>)> {
|
||||
pub(super) fn new_pair(is_nonblocking: bool) -> Result<(Arc<Endpoint>, Arc<Endpoint>)> {
|
||||
let flags = if is_nonblocking {
|
||||
StatusFlags::O_NONBLOCK
|
||||
} else {
|
||||
@ -43,26 +43,26 @@ impl Endpoint {
|
||||
})
|
||||
}
|
||||
|
||||
pub fn addr(&self) -> Option<UnixSocketAddr> {
|
||||
pub(super) fn addr(&self) -> Option<UnixSocketAddrBound> {
|
||||
self.0.addr.read().clone()
|
||||
}
|
||||
|
||||
pub fn set_addr(&self, addr: UnixSocketAddr) {
|
||||
pub(super) fn set_addr(&self, addr: UnixSocketAddrBound) {
|
||||
*self.0.addr.write() = Some(addr);
|
||||
}
|
||||
|
||||
pub fn peer_addr(&self) -> Option<UnixSocketAddr> {
|
||||
pub(super) fn peer_addr(&self) -> Option<UnixSocketAddrBound> {
|
||||
self.0.peer.upgrade().map(|peer| peer.addr()).flatten()
|
||||
}
|
||||
|
||||
pub fn is_nonblocking(&self) -> bool {
|
||||
pub(super) fn is_nonblocking(&self) -> bool {
|
||||
let reader_status = self.0.reader.is_nonblocking();
|
||||
let writer_status = self.0.writer.is_nonblocking();
|
||||
debug_assert!(reader_status == writer_status);
|
||||
reader_status
|
||||
}
|
||||
|
||||
pub fn set_nonblocking(&self, is_nonblocking: bool) -> Result<()> {
|
||||
pub(super) fn set_nonblocking(&self, is_nonblocking: bool) -> Result<()> {
|
||||
let reader_flags = self.0.reader.status_flags();
|
||||
self.0
|
||||
.reader
|
||||
@ -74,15 +74,15 @@ impl Endpoint {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn read(&self, buf: &mut [u8]) -> Result<usize> {
|
||||
pub(super) fn read(&self, buf: &mut [u8]) -> Result<usize> {
|
||||
self.0.reader.read(buf)
|
||||
}
|
||||
|
||||
pub fn write(&self, buf: &[u8]) -> Result<usize> {
|
||||
pub(super) fn write(&self, buf: &[u8]) -> Result<usize> {
|
||||
self.0.writer.write(buf)
|
||||
}
|
||||
|
||||
pub fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
|
||||
pub(super) fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
|
||||
if !self.is_connected() {
|
||||
return_errno_with_message!(Errno::ENOTCONN, "The socket is not connected.");
|
||||
}
|
||||
@ -98,11 +98,11 @@ impl Endpoint {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn is_connected(&self) -> bool {
|
||||
pub(super) fn is_connected(&self) -> bool {
|
||||
self.0.peer.upgrade().is_some()
|
||||
}
|
||||
|
||||
pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
|
||||
pub(super) fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
|
||||
let mut events = IoEvents::empty();
|
||||
// FIXME: should reader and writer use the same mask?
|
||||
let reader_events = self.0.reader.poll(mask, poller);
|
||||
|
||||
@ -1,50 +1,99 @@
|
||||
use core::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
use crate::fs::utils::{IoEvents, Pollee, Poller};
|
||||
use crate::net::socket::unix::addr::UnixSocketAddr;
|
||||
use crate::fs::fs_resolver::{split_path, FsPath};
|
||||
use crate::fs::utils::{Dentry, InodeMode, InodeType, IoEvents, Pollee, Poller};
|
||||
use crate::net::socket::unix::addr::{UnixSocketAddr, UnixSocketAddrBound};
|
||||
use crate::prelude::*;
|
||||
|
||||
pub struct Init {
|
||||
use super::connected::Connected;
|
||||
use super::endpoint::Endpoint;
|
||||
use super::listener::push_incoming;
|
||||
|
||||
pub(super) struct Init {
|
||||
is_nonblocking: AtomicBool,
|
||||
bind_addr: Option<UnixSocketAddr>,
|
||||
addr: Mutex<Option<UnixSocketAddrBound>>,
|
||||
pollee: Pollee,
|
||||
}
|
||||
|
||||
impl Init {
|
||||
pub fn new(is_nonblocking: bool) -> Self {
|
||||
pub(super) fn new(is_nonblocking: bool) -> Self {
|
||||
Self {
|
||||
is_nonblocking: AtomicBool::new(is_nonblocking),
|
||||
bind_addr: None,
|
||||
addr: Mutex::new(None),
|
||||
pollee: Pollee::new(IoEvents::empty()),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn bind(&mut self, mut addr: UnixSocketAddr) -> Result<()> {
|
||||
if self.bind_addr.is_some() {
|
||||
pub(super) fn bind(&self, addr_to_bind: &UnixSocketAddr) -> Result<()> {
|
||||
let mut addr = self.addr.lock();
|
||||
if addr.is_some() {
|
||||
return_errno_with_message!(Errno::EINVAL, "the socket is already bound");
|
||||
}
|
||||
addr.create_file_and_bind()?;
|
||||
self.bind_addr = Some(addr);
|
||||
|
||||
let bound_addr = match addr_to_bind {
|
||||
UnixSocketAddr::Abstract(_) => todo!(),
|
||||
UnixSocketAddr::Path(path) => {
|
||||
let dentry = create_socket_file(path)?;
|
||||
UnixSocketAddrBound::Path(dentry)
|
||||
}
|
||||
};
|
||||
|
||||
*addr = Some(bound_addr);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn is_bound(&self) -> bool {
|
||||
self.bind_addr.is_none()
|
||||
pub(super) fn connect(&self, remote_addr: &UnixSocketAddrBound) -> Result<Connected> {
|
||||
let addr = self.addr();
|
||||
|
||||
if let Some(ref addr) = addr {
|
||||
if *addr == *remote_addr {
|
||||
return_errno_with_message!(Errno::EINVAL, "try to connect to self is invalid");
|
||||
}
|
||||
}
|
||||
|
||||
let (this_end, remote_end) = Endpoint::new_pair(self.is_nonblocking())?;
|
||||
remote_end.set_addr(remote_addr.clone());
|
||||
if let Some(addr) = addr {
|
||||
this_end.set_addr(addr.clone());
|
||||
};
|
||||
|
||||
push_incoming(remote_addr, remote_end)?;
|
||||
Ok(Connected::new(this_end))
|
||||
}
|
||||
|
||||
pub fn bound_addr(&self) -> Option<&UnixSocketAddr> {
|
||||
self.bind_addr.as_ref()
|
||||
pub(super) fn is_bound(&self) -> bool {
|
||||
self.addr.lock().is_some()
|
||||
}
|
||||
|
||||
pub fn is_nonblocking(&self) -> bool {
|
||||
pub(super) fn addr(&self) -> Option<UnixSocketAddrBound> {
|
||||
self.addr.lock().clone()
|
||||
}
|
||||
|
||||
pub(super) fn is_nonblocking(&self) -> bool {
|
||||
self.is_nonblocking.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
pub fn set_nonblocking(&self, is_nonblocking: bool) {
|
||||
pub(super) fn set_nonblocking(&self, is_nonblocking: bool) {
|
||||
self.is_nonblocking.store(is_nonblocking, Ordering::Release);
|
||||
}
|
||||
|
||||
pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
|
||||
pub(super) fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
|
||||
self.pollee.poll(mask, poller)
|
||||
}
|
||||
}
|
||||
|
||||
fn create_socket_file(path: &str) -> Result<Arc<Dentry>> {
|
||||
let (parent_pathname, file_name) = split_path(path);
|
||||
let parent = {
|
||||
let current = current!();
|
||||
let fs = current.fs().read();
|
||||
let parent_path = FsPath::try_from(parent_pathname)?;
|
||||
fs.lookup(&parent_path)?
|
||||
};
|
||||
let dentry = parent.create(
|
||||
file_name,
|
||||
InodeType::Socket,
|
||||
InodeMode::S_IRUSR | InodeMode::S_IWUSR,
|
||||
)?;
|
||||
Ok(dentry)
|
||||
}
|
||||
|
||||
@ -1,29 +0,0 @@
|
||||
use core::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
use crate::net::socket::unix::addr::UnixSocketAddr;
|
||||
|
||||
pub struct Listen {
|
||||
addr: UnixSocketAddr,
|
||||
is_nonblocking: AtomicBool,
|
||||
}
|
||||
|
||||
impl Listen {
|
||||
pub fn new(addr: UnixSocketAddr, nonblocking: bool) -> Self {
|
||||
Self {
|
||||
addr,
|
||||
is_nonblocking: AtomicBool::new(nonblocking),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn addr(&self) -> &UnixSocketAddr {
|
||||
&self.addr
|
||||
}
|
||||
|
||||
pub fn is_nonblocking(&self) -> bool {
|
||||
self.is_nonblocking.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
pub fn set_nonblocking(&self, is_nonblocking: bool) {
|
||||
self.is_nonblocking.store(is_nonblocking, Ordering::Release);
|
||||
}
|
||||
}
|
||||
@ -1,116 +1,171 @@
|
||||
use super::{connected::Connected, endpoint::Endpoint, UnixStreamSocket};
|
||||
use crate::fs::file_handle::FileLike;
|
||||
use crate::fs::utils::{Dentry, Inode, IoEvents, Pollee, Poller};
|
||||
use crate::net::socket::unix::addr::{UnixSocketAddr, UnixSocketAddrBound};
|
||||
use crate::net::socket::SocketAddr;
|
||||
use crate::prelude::*;
|
||||
use core::sync::atomic::{AtomicBool, Ordering};
|
||||
use keyable_arc::KeyableWeak;
|
||||
use spin::RwLockReadGuard;
|
||||
|
||||
use crate::{
|
||||
fs::utils::{Inode, IoEvents, Pollee, Poller},
|
||||
net::socket::unix::addr::UnixSocketAddr,
|
||||
prelude::*,
|
||||
};
|
||||
pub(super) struct Listener {
|
||||
addr: UnixSocketAddrBound,
|
||||
is_nonblocking: AtomicBool,
|
||||
}
|
||||
|
||||
use super::endpoint::Endpoint;
|
||||
impl Listener {
|
||||
pub(super) fn new(
|
||||
addr: UnixSocketAddrBound,
|
||||
backlog: usize,
|
||||
nonblocking: bool,
|
||||
) -> Result<Self> {
|
||||
BACKLOG_TABLE.add_backlog(&addr, backlog)?;
|
||||
Ok(Self {
|
||||
addr,
|
||||
is_nonblocking: AtomicBool::new(nonblocking),
|
||||
})
|
||||
}
|
||||
|
||||
pub static ACTIVE_LISTENERS: ActiveListeners = ActiveListeners::new();
|
||||
pub(super) fn addr(&self) -> &UnixSocketAddrBound {
|
||||
&self.addr
|
||||
}
|
||||
|
||||
pub struct ActiveListeners {
|
||||
listeners: RwLock<BTreeMap<KeyableWeak<dyn Inode>, Arc<Listener>>>,
|
||||
pub(super) fn is_nonblocking(&self) -> bool {
|
||||
self.is_nonblocking.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
pub(super) fn set_nonblocking(&self, is_nonblocking: bool) {
|
||||
self.is_nonblocking.store(is_nonblocking, Ordering::Release);
|
||||
}
|
||||
|
||||
pub(super) fn accept(&self) -> Result<(Arc<dyn FileLike>, SocketAddr)> {
|
||||
let addr = self.addr().clone();
|
||||
let is_nonblocking = self.is_nonblocking();
|
||||
|
||||
let connected = {
|
||||
let local_endpoint = BACKLOG_TABLE.pop_incoming(is_nonblocking, &addr)?;
|
||||
Connected::new(local_endpoint)
|
||||
};
|
||||
|
||||
let peer_addr = match connected.peer_addr() {
|
||||
None => SocketAddr::Unix(UnixSocketAddr::Path(String::new())),
|
||||
Some(addr) => SocketAddr::from(addr.clone()),
|
||||
};
|
||||
|
||||
let socket = Arc::new(UnixStreamSocket::new_connected(connected));
|
||||
|
||||
Ok((socket, peer_addr))
|
||||
}
|
||||
|
||||
pub(super) fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
|
||||
let addr = self.addr();
|
||||
let backlog = BACKLOG_TABLE.get_backlog(addr).unwrap();
|
||||
backlog.poll(mask, poller)
|
||||
}
|
||||
}
|
||||
|
||||
static BACKLOG_TABLE: BacklogTable = BacklogTable::new();
|
||||
|
||||
struct BacklogTable {
|
||||
backlog_sockets: RwLock<BTreeMap<KeyableWeak<dyn Inode>, Arc<Backlog>>>,
|
||||
// TODO: For linux, there is also abstract socket domain that a socket addr is not bound to an inode.
|
||||
}
|
||||
|
||||
impl ActiveListeners {
|
||||
pub const fn new() -> Self {
|
||||
impl BacklogTable {
|
||||
const fn new() -> Self {
|
||||
Self {
|
||||
listeners: RwLock::new(BTreeMap::new()),
|
||||
backlog_sockets: RwLock::new(BTreeMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn add_listener(&self, addr: &UnixSocketAddr, backlog: usize) -> Result<()> {
|
||||
let inode = create_keyable_inode(addr)?;
|
||||
let mut listeners = self.listeners.write();
|
||||
if listeners.contains_key(&inode) {
|
||||
fn add_backlog(&self, addr: &UnixSocketAddrBound, backlog: usize) -> Result<()> {
|
||||
let inode = {
|
||||
let UnixSocketAddrBound::Path(dentry) = addr else {
|
||||
todo!()
|
||||
};
|
||||
create_keyable_inode(dentry)
|
||||
};
|
||||
|
||||
let mut backlog_sockets = self.backlog_sockets.write();
|
||||
if backlog_sockets.contains_key(&inode) {
|
||||
return_errno_with_message!(Errno::EADDRINUSE, "the addr is already used");
|
||||
}
|
||||
let new_listener = Arc::new(Listener::new(backlog));
|
||||
listeners.insert(inode, new_listener);
|
||||
let new_backlog = Arc::new(Backlog::new(backlog));
|
||||
backlog_sockets.insert(inode, new_backlog);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) fn get_listener(&self, addr: &UnixSocketAddr) -> Result<Arc<Listener>> {
|
||||
let listeners = self.listeners.read();
|
||||
get_listener(&listeners, addr)
|
||||
fn get_backlog(&self, addr: &UnixSocketAddrBound) -> Result<Arc<Backlog>> {
|
||||
let inode = {
|
||||
let UnixSocketAddrBound::Path(dentry) = addr else {
|
||||
todo!()
|
||||
};
|
||||
create_keyable_inode(dentry)
|
||||
};
|
||||
|
||||
let backlog_sockets = self.backlog_sockets.read();
|
||||
backlog_sockets
|
||||
.get(&inode)
|
||||
.map(Arc::clone)
|
||||
.ok_or_else(|| Error::with_message(Errno::EINVAL, "the socket is not listened"))
|
||||
}
|
||||
|
||||
pub(super) fn pop_incoming(
|
||||
&self,
|
||||
nonblocking: bool,
|
||||
addr: &UnixSocketAddr,
|
||||
) -> Result<Arc<Endpoint>> {
|
||||
fn pop_incoming(&self, nonblocking: bool, addr: &UnixSocketAddrBound) -> Result<Arc<Endpoint>> {
|
||||
let poller = Poller::new();
|
||||
loop {
|
||||
let listener = {
|
||||
let listeners = self.listeners.read();
|
||||
get_listener(&listeners, addr)?
|
||||
};
|
||||
if let Some(endpoint) = listener.pop_incoming() {
|
||||
let backlog = self.get_backlog(addr)?;
|
||||
|
||||
if let Some(endpoint) = backlog.pop_incoming() {
|
||||
return Ok(endpoint);
|
||||
}
|
||||
|
||||
if nonblocking {
|
||||
return_errno_with_message!(Errno::EAGAIN, "no connection comes");
|
||||
}
|
||||
|
||||
let events = {
|
||||
let mask = IoEvents::IN;
|
||||
listener.poll(mask, Some(&poller))
|
||||
backlog.poll(mask, Some(&poller))
|
||||
};
|
||||
|
||||
if events.contains(IoEvents::ERR) | events.contains(IoEvents::HUP) {
|
||||
return_errno_with_message!(Errno::EINVAL, "connection is refused");
|
||||
return_errno_with_message!(Errno::ECONNABORTED, "connection is aborted");
|
||||
}
|
||||
|
||||
if events.is_empty() {
|
||||
poller.wait();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn push_incoming(
|
||||
&self,
|
||||
addr: &UnixSocketAddr,
|
||||
endpoint: Arc<Endpoint>,
|
||||
) -> Result<()> {
|
||||
let listeners = self.listeners.read();
|
||||
let listener = get_listener(&listeners, addr).map_err(|_| {
|
||||
fn push_incoming(&self, addr: &UnixSocketAddrBound, endpoint: Arc<Endpoint>) -> Result<()> {
|
||||
let backlog = self.get_backlog(addr).map_err(|_| {
|
||||
Error::with_message(
|
||||
Errno::ECONNREFUSED,
|
||||
"no socket is listened at the remote address",
|
||||
)
|
||||
})?;
|
||||
listener.push_incoming(endpoint)
|
||||
|
||||
backlog.push_incoming(endpoint)
|
||||
}
|
||||
|
||||
pub(super) fn remove_listener(&self, addr: &UnixSocketAddr) {
|
||||
let Ok(inode) = create_keyable_inode(addr) else {
|
||||
return;
|
||||
fn remove_backlog(&self, addr: &UnixSocketAddrBound) {
|
||||
let UnixSocketAddrBound::Path(dentry) = addr else {
|
||||
todo!()
|
||||
};
|
||||
self.listeners.write().remove(&inode);
|
||||
|
||||
let inode = create_keyable_inode(dentry);
|
||||
self.backlog_sockets.write().remove(&inode);
|
||||
}
|
||||
}
|
||||
|
||||
fn get_listener(
|
||||
listeners: &RwLockReadGuard<BTreeMap<KeyableWeak<dyn Inode>, Arc<Listener>>>,
|
||||
addr: &UnixSocketAddr,
|
||||
) -> Result<Arc<Listener>> {
|
||||
let dentry = create_keyable_inode(addr)?;
|
||||
listeners
|
||||
.get(&dentry)
|
||||
.map(Arc::clone)
|
||||
.ok_or_else(|| Error::with_message(Errno::EINVAL, "the socket is not listened"))
|
||||
}
|
||||
|
||||
pub(super) struct Listener {
|
||||
struct Backlog {
|
||||
pollee: Pollee,
|
||||
backlog: usize,
|
||||
incoming_endpoints: Mutex<VecDeque<Arc<Endpoint>>>,
|
||||
}
|
||||
|
||||
impl Listener {
|
||||
pub fn new(backlog: usize) -> Self {
|
||||
impl Backlog {
|
||||
fn new(backlog: usize) -> Self {
|
||||
Self {
|
||||
pollee: Pollee::new(IoEvents::empty()),
|
||||
backlog,
|
||||
@ -118,7 +173,7 @@ impl Listener {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn push_incoming(&self, endpoint: Arc<Endpoint>) -> Result<()> {
|
||||
fn push_incoming(&self, endpoint: Arc<Endpoint>) -> Result<()> {
|
||||
let mut endpoints = self.incoming_endpoints.lock();
|
||||
if endpoints.len() >= self.backlog {
|
||||
return_errno_with_message!(Errno::ECONNREFUSED, "incoming_endpoints is full");
|
||||
@ -128,24 +183,34 @@ impl Listener {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn pop_incoming(&self) -> Option<Arc<Endpoint>> {
|
||||
fn pop_incoming(&self) -> Option<Arc<Endpoint>> {
|
||||
let mut incoming_endpoints = self.incoming_endpoints.lock();
|
||||
let endpoint = incoming_endpoints.pop_front();
|
||||
if endpoint.is_none() {
|
||||
if incoming_endpoints.is_empty() {
|
||||
self.pollee.del_events(IoEvents::IN);
|
||||
}
|
||||
endpoint
|
||||
}
|
||||
|
||||
pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
|
||||
fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
|
||||
// Lock to avoid any events may change pollee state when we poll
|
||||
let _lock = self.incoming_endpoints.lock();
|
||||
self.pollee.poll(mask, poller)
|
||||
}
|
||||
}
|
||||
|
||||
fn create_keyable_inode(addr: &UnixSocketAddr) -> Result<KeyableWeak<dyn Inode>> {
|
||||
let dentry = addr.dentry()?;
|
||||
fn create_keyable_inode(dentry: &Arc<Dentry>) -> KeyableWeak<dyn Inode> {
|
||||
let inode = dentry.inode();
|
||||
Ok(KeyableWeak::from(inode))
|
||||
KeyableWeak::from(inode)
|
||||
}
|
||||
|
||||
pub(super) fn unregister_backlog(addr: &UnixSocketAddrBound) {
|
||||
BACKLOG_TABLE.remove_backlog(&addr);
|
||||
}
|
||||
|
||||
pub(super) fn push_incoming(
|
||||
remote_addr: &UnixSocketAddrBound,
|
||||
remote_end: Arc<Endpoint>,
|
||||
) -> Result<()> {
|
||||
BACKLOG_TABLE.push_incoming(remote_addr, remote_end)
|
||||
}
|
||||
|
||||
@ -1,9 +1,7 @@
|
||||
mod connected;
|
||||
mod endpoint;
|
||||
mod init;
|
||||
mod listen;
|
||||
mod listener;
|
||||
pub mod stream;
|
||||
mod stream;
|
||||
|
||||
pub use listener::{ActiveListeners, ACTIVE_LISTENERS};
|
||||
pub use stream::UnixStreamSocket;
|
||||
|
||||
@ -1,6 +1,8 @@
|
||||
use crate::fs::file_handle::FileLike;
|
||||
use crate::fs::utils::{IoEvents, Poller, StatusFlags};
|
||||
use crate::net::socket::unix::addr::UnixSocketAddr;
|
||||
use crate::fs::fs_resolver::FsPath;
|
||||
use crate::fs::utils::{Dentry, InodeType, IoEvents, Poller, StatusFlags};
|
||||
use crate::net::socket::unix::addr::UnixSocketAddrBound;
|
||||
use crate::net::socket::unix::UnixSocketAddr;
|
||||
use crate::net::socket::util::send_recv_flags::SendRecvFlags;
|
||||
use crate::net::socket::util::sockaddr::SocketAddr;
|
||||
use crate::net::socket::{SockShutdownCmd, Socket};
|
||||
@ -9,40 +11,59 @@ use crate::prelude::*;
|
||||
use super::connected::Connected;
|
||||
use super::endpoint::Endpoint;
|
||||
use super::init::Init;
|
||||
use super::listen::Listen;
|
||||
use super::ACTIVE_LISTENERS;
|
||||
use super::listener::{unregister_backlog, Listener};
|
||||
|
||||
pub struct UnixStreamSocket(RwLock<Status>);
|
||||
pub struct UnixStreamSocket(RwLock<State>);
|
||||
|
||||
enum Status {
|
||||
Init(Init),
|
||||
Listen(Listen),
|
||||
Connected(Connected),
|
||||
impl UnixStreamSocket {
|
||||
pub(super) fn new_init(init: Init) -> Self {
|
||||
Self(RwLock::new(State::Init(Arc::new(init))))
|
||||
}
|
||||
|
||||
pub(super) fn new_listen(listen: Listener) -> Self {
|
||||
Self(RwLock::new(State::Listen(Arc::new(listen))))
|
||||
}
|
||||
|
||||
pub(super) fn new_connected(connected: Connected) -> Self {
|
||||
Self(RwLock::new(State::Connected(Arc::new(connected))))
|
||||
}
|
||||
}
|
||||
|
||||
enum State {
|
||||
Init(Arc<Init>),
|
||||
Listen(Arc<Listener>),
|
||||
Connected(Arc<Connected>),
|
||||
}
|
||||
|
||||
impl UnixStreamSocket {
|
||||
pub fn new(nonblocking: bool) -> Self {
|
||||
let status = Status::Init(Init::new(nonblocking));
|
||||
Self(RwLock::new(status))
|
||||
let init = Init::new(nonblocking);
|
||||
Self::new_init(init)
|
||||
}
|
||||
|
||||
pub fn new_pair(nonblocking: bool) -> Result<(Arc<Self>, Arc<Self>)> {
|
||||
let (end_a, end_b) = Endpoint::end_pair(nonblocking)?;
|
||||
let connected_a = UnixStreamSocket(RwLock::new(Status::Connected(Connected::new(end_a))));
|
||||
let connected_b = UnixStreamSocket(RwLock::new(Status::Connected(Connected::new(end_b))));
|
||||
let (end_a, end_b) = Endpoint::new_pair(nonblocking)?;
|
||||
let connected_a = {
|
||||
let connected = Connected::new(end_a);
|
||||
Self::new_connected(connected)
|
||||
};
|
||||
let connected_b = {
|
||||
let connected = Connected::new(end_b);
|
||||
Self::new_connected(connected)
|
||||
};
|
||||
Ok((Arc::new(connected_a), Arc::new(connected_b)))
|
||||
}
|
||||
|
||||
fn bound_addr(&self) -> Option<UnixSocketAddr> {
|
||||
fn bound_addr(&self) -> Option<UnixSocketAddrBound> {
|
||||
let status = self.0.read();
|
||||
match &*status {
|
||||
Status::Init(init) => init.bound_addr().map(Clone::clone),
|
||||
Status::Listen(listen) => Some(listen.addr().clone()),
|
||||
Status::Connected(connected) => connected.addr(),
|
||||
State::Init(init) => init.addr(),
|
||||
State::Listen(listen) => Some(listen.addr().clone()),
|
||||
State::Connected(connected) => connected.addr(),
|
||||
}
|
||||
}
|
||||
|
||||
fn supported_flags(status_flags: &StatusFlags) -> StatusFlags {
|
||||
fn mask_flags(status_flags: &StatusFlags) -> StatusFlags {
|
||||
const SUPPORTED_FLAGS: StatusFlags = StatusFlags::O_NONBLOCK;
|
||||
const UNSUPPORTED_FLAGS: StatusFlags = SUPPORTED_FLAGS.complement();
|
||||
|
||||
@ -71,22 +92,18 @@ impl FileLike for UnixStreamSocket {
|
||||
fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
|
||||
let inner = self.0.read();
|
||||
match &*inner {
|
||||
Status::Init(init) => init.poll(mask, poller),
|
||||
Status::Listen(listen) => {
|
||||
let addr = listen.addr();
|
||||
let listener = ACTIVE_LISTENERS.get_listener(addr).unwrap();
|
||||
listener.poll(mask, poller)
|
||||
}
|
||||
Status::Connected(connet) => todo!(),
|
||||
State::Init(init) => init.poll(mask, poller),
|
||||
State::Listen(listen) => listen.poll(mask, poller),
|
||||
State::Connected(connected) => connected.poll(mask, poller),
|
||||
}
|
||||
}
|
||||
|
||||
fn status_flags(&self) -> StatusFlags {
|
||||
let inner = self.0.read();
|
||||
let is_nonblocking = match &*inner {
|
||||
Status::Init(init) => init.is_nonblocking(),
|
||||
Status::Listen(listen) => listen.is_nonblocking(),
|
||||
Status::Connected(connected) => connected.is_nonblocking(),
|
||||
State::Init(init) => init.is_nonblocking(),
|
||||
State::Listen(listen) => listen.is_nonblocking(),
|
||||
State::Connected(connected) => connected.is_nonblocking(),
|
||||
};
|
||||
|
||||
if is_nonblocking {
|
||||
@ -98,15 +115,15 @@ impl FileLike for UnixStreamSocket {
|
||||
|
||||
fn set_status_flags(&self, new_flags: StatusFlags) -> Result<()> {
|
||||
let is_nonblocking = {
|
||||
let supported_flags = Self::supported_flags(&new_flags);
|
||||
let supported_flags = Self::mask_flags(&new_flags);
|
||||
supported_flags.contains(StatusFlags::O_NONBLOCK)
|
||||
};
|
||||
|
||||
let mut inner = self.0.write();
|
||||
match &mut *inner {
|
||||
Status::Init(init) => init.set_nonblocking(is_nonblocking),
|
||||
Status::Listen(listen) => listen.set_nonblocking(is_nonblocking),
|
||||
Status::Connected(connected) => connected.set_nonblocking(is_nonblocking),
|
||||
State::Init(init) => init.set_nonblocking(is_nonblocking),
|
||||
State::Listen(listen) => listen.set_nonblocking(is_nonblocking),
|
||||
State::Connected(connected) => connected.set_nonblocking(is_nonblocking),
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@ -115,114 +132,93 @@ impl FileLike for UnixStreamSocket {
|
||||
impl Socket for UnixStreamSocket {
|
||||
fn bind(&self, sockaddr: SocketAddr) -> Result<()> {
|
||||
let addr = UnixSocketAddr::try_from(sockaddr)?;
|
||||
let mut inner = self.0.write();
|
||||
match &mut *inner {
|
||||
Status::Init(init) => init.bind(addr),
|
||||
Status::Listen(_) | Status::Connected(_) => {
|
||||
return_errno_with_message!(
|
||||
Errno::EINVAL,
|
||||
"cannot bind a listening or connected socket"
|
||||
);
|
||||
} // FIXME: Maybe binding a connected sockted should also be allowed?
|
||||
}
|
||||
|
||||
let init = match &*self.0.read() {
|
||||
State::Init(init) => init.clone(),
|
||||
_ => return_errno_with_message!(
|
||||
Errno::EINVAL,
|
||||
"cannot bind a listening or connected socket"
|
||||
),
|
||||
// FIXME: Maybe binding a connected socket should also be allowed?
|
||||
};
|
||||
|
||||
init.bind(&addr)
|
||||
}
|
||||
|
||||
fn connect(&self, sockaddr: SocketAddr) -> Result<()> {
|
||||
let mut inner = self.0.write();
|
||||
match &*inner {
|
||||
Status::Init(init) => {
|
||||
let remote_addr = UnixSocketAddr::try_from(sockaddr)?;
|
||||
let addr = init.bound_addr();
|
||||
if let Some(addr) = addr {
|
||||
if addr.path() == remote_addr.path() {
|
||||
return_errno_with_message!(
|
||||
Errno::EINVAL,
|
||||
"try to connect to self is invalid"
|
||||
);
|
||||
}
|
||||
let remote_addr = {
|
||||
let unix_socket_addr = UnixSocketAddr::try_from(sockaddr)?;
|
||||
match unix_socket_addr {
|
||||
UnixSocketAddr::Abstract(abstract_name) => {
|
||||
UnixSocketAddrBound::Abstract(abstract_name)
|
||||
}
|
||||
UnixSocketAddr::Path(path) => {
|
||||
let dentry = lookup_socket_file(&path)?;
|
||||
UnixSocketAddrBound::Path(dentry)
|
||||
}
|
||||
let (this_end, remote_end) = Endpoint::end_pair(init.is_nonblocking())?;
|
||||
remote_end.set_addr(remote_addr.clone());
|
||||
if let Some(addr) = addr {
|
||||
this_end.set_addr(addr.clone());
|
||||
};
|
||||
ACTIVE_LISTENERS.push_incoming(&remote_addr, remote_end)?;
|
||||
*inner = Status::Connected(Connected::new(this_end));
|
||||
Ok(())
|
||||
}
|
||||
Status::Listen(_) => {
|
||||
return_errno_with_message!(Errno::EINVAL, "the socket is listened")
|
||||
}
|
||||
Status::Connected(_) => {
|
||||
};
|
||||
|
||||
let init = match &*self.0.read() {
|
||||
State::Init(init) => init.clone(),
|
||||
State::Listen(_) => return_errno_with_message!(Errno::EINVAL, "the socket is listened"),
|
||||
State::Connected(_) => {
|
||||
return_errno_with_message!(Errno::EISCONN, "the socket is connected")
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let connected = init.connect(&remote_addr)?;
|
||||
|
||||
*self.0.write() = State::Connected(Arc::new(connected));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn listen(&self, backlog: usize) -> Result<()> {
|
||||
let mut inner = self.0.write();
|
||||
match &*inner {
|
||||
Status::Init(init) => {
|
||||
let addr = init.bound_addr().ok_or(Error::with_message(
|
||||
Errno::EINVAL,
|
||||
"the socket is not bound",
|
||||
))?;
|
||||
ACTIVE_LISTENERS.add_listener(addr, backlog)?;
|
||||
*inner = Status::Listen(Listen::new(addr.clone(), init.is_nonblocking()));
|
||||
return Ok(());
|
||||
let init = match &*self.0.read() {
|
||||
State::Init(init) => init.clone(),
|
||||
State::Listen(_) => {
|
||||
return_errno_with_message!(Errno::EINVAL, "the socket is already listening")
|
||||
}
|
||||
Status::Listen(_) => {
|
||||
return_errno_with_message!(Errno::EINVAL, "the socket is already listened")
|
||||
}
|
||||
Status::Connected(_) => {
|
||||
return_errno_with_message!(Errno::EINVAL, "the socket is already connected")
|
||||
State::Connected(_) => {
|
||||
return_errno_with_message!(Errno::EISCONN, "the socket is already connected")
|
||||
}
|
||||
};
|
||||
|
||||
let addr = init.addr().ok_or(Error::with_message(
|
||||
Errno::EINVAL,
|
||||
"the socket is not bound",
|
||||
))?;
|
||||
|
||||
let listener = Listener::new(addr.clone(), backlog, init.is_nonblocking())?;
|
||||
*self.0.write() = State::Listen(Arc::new(listener));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn accept(&self) -> Result<(Arc<dyn FileLike>, SocketAddr)> {
|
||||
let inner = self.0.read();
|
||||
match &*inner {
|
||||
Status::Listen(listen) => {
|
||||
let is_nonblocking = listen.is_nonblocking();
|
||||
let addr = listen.addr().clone();
|
||||
drop(inner);
|
||||
// Avoid lock when waiting
|
||||
let connected = {
|
||||
let local_endpoint = ACTIVE_LISTENERS.pop_incoming(is_nonblocking, &addr)?;
|
||||
Connected::new(local_endpoint)
|
||||
};
|
||||
let listen = match &*self.0.read() {
|
||||
State::Listen(listen) => listen.clone(),
|
||||
_ => return_errno_with_message!(Errno::EINVAL, "the socket is not listening"),
|
||||
};
|
||||
|
||||
let peer_addr = match connected.peer_addr() {
|
||||
None => SocketAddr::Unix(String::new()),
|
||||
Some(addr) => SocketAddr::from(addr.clone()),
|
||||
};
|
||||
|
||||
let socket = UnixStreamSocket(RwLock::new(Status::Connected(connected)));
|
||||
return Ok((Arc::new(socket), peer_addr));
|
||||
}
|
||||
Status::Connected(_) | Status::Init(_) => {
|
||||
return_errno_with_message!(Errno::EINVAL, "the socket is not listened")
|
||||
}
|
||||
}
|
||||
listen.accept()
|
||||
}
|
||||
|
||||
fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
|
||||
let inner = self.0.read();
|
||||
if let Status::Connected(connected) = &*inner {
|
||||
connected.shutdown(cmd)
|
||||
} else {
|
||||
return_errno_with_message!(Errno::ENOTCONN, "the socked is not connected");
|
||||
}
|
||||
let connected = match &*self.0.read() {
|
||||
State::Connected(connected) => connected.clone(),
|
||||
_ => return_errno_with_message!(Errno::ENOTCONN, "the socked is not connected"),
|
||||
};
|
||||
|
||||
connected.shutdown(cmd)
|
||||
}
|
||||
|
||||
fn addr(&self) -> Result<SocketAddr> {
|
||||
let inner = self.0.read();
|
||||
let addr = match &*inner {
|
||||
Status::Init(init) => init.bound_addr().map(Clone::clone),
|
||||
Status::Listen(listen) => Some(listen.addr().clone()),
|
||||
Status::Connected(connected) => connected.addr(),
|
||||
let addr = match &*self.0.read() {
|
||||
State::Init(init) => init.addr(),
|
||||
State::Listen(listen) => Some(listen.addr().clone()),
|
||||
State::Connected(connected) => connected.addr(),
|
||||
};
|
||||
|
||||
addr.map(Into::<SocketAddr>::into)
|
||||
.ok_or(Error::with_message(
|
||||
Errno::EINVAL,
|
||||
@ -231,31 +227,28 @@ impl Socket for UnixStreamSocket {
|
||||
}
|
||||
|
||||
fn peer_addr(&self) -> Result<SocketAddr> {
|
||||
let inner = self.0.read();
|
||||
if let Status::Connected(connected) = &*inner {
|
||||
match connected.peer_addr() {
|
||||
None => return Ok(SocketAddr::Unix(String::new())),
|
||||
Some(peer_addr) => {
|
||||
return Ok(SocketAddr::from(peer_addr.clone()));
|
||||
}
|
||||
let connected = match &*self.0.read() {
|
||||
State::Connected(connected) => connected.clone(),
|
||||
_ => return_errno_with_message!(Errno::ENOTCONN, "the socket is not connected"),
|
||||
};
|
||||
|
||||
match connected.peer_addr() {
|
||||
None => return Ok(SocketAddr::Unix(UnixSocketAddr::Path(String::new()))),
|
||||
Some(peer_addr) => {
|
||||
return Ok(SocketAddr::from(peer_addr.clone()));
|
||||
}
|
||||
}
|
||||
return_errno_with_message!(Errno::EINVAL, "the socket is not connected");
|
||||
}
|
||||
|
||||
fn recvfrom(&self, buf: &mut [u8], flags: SendRecvFlags) -> Result<(usize, SocketAddr)> {
|
||||
let inner = self.0.read();
|
||||
// TODO: deal with flags
|
||||
match &*inner {
|
||||
Status::Connected(connected) => {
|
||||
let read_size = connected.read(buf)?;
|
||||
let peer_addr = self.peer_addr()?;
|
||||
Ok((read_size, peer_addr))
|
||||
}
|
||||
Status::Init(_) | Status::Listen(_) => {
|
||||
return_errno_with_message!(Errno::EINVAL, "the socket is not connected")
|
||||
}
|
||||
}
|
||||
let connected = match &*self.0.read() {
|
||||
State::Connected(connected) => connected.clone(),
|
||||
_ => return_errno_with_message!(Errno::ENOTCONN, "the socket is not connected"),
|
||||
};
|
||||
|
||||
let peer_addr = self.peer_addr()?;
|
||||
let read_size = connected.read(buf)?;
|
||||
Ok((read_size, peer_addr))
|
||||
}
|
||||
|
||||
fn sendto(
|
||||
@ -266,13 +259,12 @@ impl Socket for UnixStreamSocket {
|
||||
) -> Result<usize> {
|
||||
debug_assert!(remote.is_none());
|
||||
// TODO: deal with flags
|
||||
let inner = self.0.read();
|
||||
match &*inner {
|
||||
Status::Connected(connected) => connected.write(buf),
|
||||
Status::Init(_) | Status::Listen(_) => {
|
||||
return_errno_with_message!(Errno::EINVAL, "the socket is not connected")
|
||||
}
|
||||
}
|
||||
let connected = match &*self.0.read() {
|
||||
State::Connected(connected) => connected.clone(),
|
||||
_ => return_errno_with_message!(Errno::ENOTCONN, "the socket is not connected"),
|
||||
};
|
||||
|
||||
connected.write(buf)
|
||||
}
|
||||
}
|
||||
|
||||
@ -282,6 +274,26 @@ impl Drop for UnixStreamSocket {
|
||||
return;
|
||||
};
|
||||
|
||||
ACTIVE_LISTENERS.remove_listener(&bound_addr);
|
||||
if let State::Listen(_) = &*self.0.read() {
|
||||
unregister_backlog(&bound_addr);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn lookup_socket_file(path: &str) -> Result<Arc<Dentry>> {
|
||||
let dentry = {
|
||||
let current = current!();
|
||||
let fs = current.fs().read();
|
||||
let fs_path = FsPath::try_from(path)?;
|
||||
fs.lookup(&fs_path)?
|
||||
};
|
||||
|
||||
if dentry.inode_type() != InodeType::Socket {
|
||||
return_errno_with_message!(Errno::ENOTSOCK, "not a socket file")
|
||||
}
|
||||
|
||||
if !dentry.inode_mode().is_readable() || !dentry.inode_mode().is_writable() {
|
||||
return_errno_with_message!(Errno::EACCES, "the socket cannot be read or written")
|
||||
}
|
||||
return Ok(dentry);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user