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

Merge branch 'uds-hanging' into 'main'

Browse Source 
Unblock connect(), allow getpeername while connecting

See merge request redox-os/base!91
This commit is contained in:
Jeremy Soller
2026-01-15 14:25:41 -07:00
parent a558a9e0ce 58cdb72051
commit 41659cce0f
1 changed files with 64 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ipcd/src/uds/stream.rs
+64 -63
View File
@@ -260,7 +260,7 @@ impl Socket {
))
}
fn establish(&mut self, peer: usize) -> Result<()> {
fn establish(&mut self, new_peer: usize, peer: usize) -> Result<()> {
if self.state != State::Connecting {
eprintln!(
"establish(id: {}): Cannot establish connection in state: {:?}",
@@ -269,36 +269,43 @@ impl Socket {
return Err(Error::new(EINVAL));
}
self.state = State::Accepted;
self.connection = Some(Connection::new(peer));
if let Some(conn) = &mut self.connection {
if conn.peer != peer {
// client is expecting other connection
return Err(Error::new(EAGAIN));
}
conn.peer = new_peer;
} else {
// client is dead
return Err(Error::new(EAGAIN));
}
Ok(())
}
fn connect(&mut self, other: usize, flags: usize) -> Result<()> {
fn connect(&mut self, other: &mut Socket) -> Result<()> {
match self.state {
State::Unbound | State::Bound => {
// If the socket is unbound or bound, wait for the listener to start listening.
if flags & O_NONBLOCK == O_NONBLOCK {
self.awaiting.push_back(other);
Ok(())
} else {
if other.flags & O_NONBLOCK != O_NONBLOCK {
// If the connecting target is not a listening,
// the connecting socket will block until the socket
// is ready to accept.
Err(Error::new(EWOULDBLOCK))
return Err(Error::new(EWOULDBLOCK));
}
}
State::Listening => {
// If the socket is already listening, it can accept connections.
self.awaiting.push_back(other);
Ok(())
}
_ => Err(Error::new(ECONNREFUSED)),
_ => return Err(Error::new(ECONNREFUSED)),
}
self.connect_unchecked(other);
Ok(())
}
// For socketpair, add the peer's id to the base socket's awaiting queue.
fn connect_socketpair(&mut self, peer: usize) {
self.awaiting.push_back(peer);
fn connect_unchecked(&mut self, other: &mut Socket) {
self.awaiting.push_back(other.primary_id);
other.state = State::Connecting;
other.connection = Some(Connection::new(self.primary_id));
}
fn is_listening(&self) -> bool {
@@ -405,14 +412,10 @@ impl<'sock> UdsStreamScheme<'sock> {
self.sockets.get(&id).ok_or(Error::new(EBADF))
}
fn get_connected_peer(
&self,
id: usize,
flags: MsgFlags,
) -> Result<(usize, Rc<RefCell<Socket>>), Error> {
fn get_connected_peer(&self, id: usize) -> Result<(usize, Rc<RefCell<Socket>>), Error> {
let mut socket = self.get_socket(id)?.borrow_mut();
let remote_id = socket.require_connected_connection(flags)?.peer;
let remote_id = socket.require_connection()?.peer;
let remote_rc = self.get_socket(remote_id).map_err(|e| {
eprintln!("get_connected_peer(id: {}): Peer socket (id: {}) has vanished. Original error: {:?}", id, remote_id, e);
Error::new(EPIPE)
@@ -537,14 +540,15 @@ impl<'sock> UdsStreamScheme<'sock> {
//
// Phase 1: The listener is bound but not yet listening.
// The client is trying to connect.
// If the listener is not listening, the client will block
// and wait until the listener starts listening.
// If the listener is not listening, the listener will
// refuse to connect until the listener starts listening.
//
// Phase 2: The listener is now listening.
// The client is still trying to connect.
// The client pushes its ID to the listener's awaiting queue
// and sets its state to `Connecting`.
// The client then blocks, waiting for the listener to accept it.
// The client will be blocked from sending/receiving messages,
// waiting for the listener to accept it.
//
// Phase 3: The listener accepts the client, changes its state to `Established`,
// and then changes the client's state to `Accepted`.
@@ -552,12 +556,6 @@ impl<'sock> UdsStreamScheme<'sock> {
// and changes its own state to `Established`.
//
// After these three phases, the socket connection is considered established.
//
// After these three phases, the socket connection is considered established.
//
// The reason why `connect` is complicated is that if the processing blocks,
// the SQE will be pushed back to the scheme's request queue,
// and the same SQE will be woken up later.
fn handle_connect(&mut self, id: usize, token_buf: &[u8]) -> Result<usize> {
let token = read_num::<u64>(token_buf)?;
let (listener_id, flags) = {
@@ -570,11 +568,20 @@ impl<'sock> UdsStreamScheme<'sock> {
let client_rc = self.get_socket(id)?.clone();
let mut client = client_rc.borrow_mut();
// Phase 1: listener is bound but not yet listening
let mut listener = listener_rc.borrow_mut();
let listener_id = listener.primary_id;
match client.state {
State::Connecting => {
// If the client is already Connecting
// Fence to prevent calling connect multiple times.
return Err(Error::new(EWOULDBLOCK));
if client
.connection
.as_ref()
.is_some_and(|c| c.peer == listener_id)
{
// No op
return Ok(0);
}
}
State::Established => {
return Err(Error::new(EISCONN));
@@ -587,13 +594,8 @@ impl<'sock> UdsStreamScheme<'sock> {
_ => {}
}
// Phase 1: listener is bound but not yet listening
let mut listener = listener_rc.borrow_mut();
let listener_id = listener.primary_id;
listener.connect(id, client.flags)?;
// Phase 2: listener is now listening
client.state = State::Connecting;
listener.connect(&mut client)?;
(listener_id, client.flags)
};
@@ -601,13 +603,6 @@ impl<'sock> UdsStreamScheme<'sock> {
// client, we'll do the same too (but also readable, why
// not)
self.post_fevent(listener_id, EVENT_READ | EVENT_WRITE)?;
// Blocking pattern
if flags & O_NONBLOCK == 0 {
return Err(Error::new(EWOULDBLOCK));
}
// Non-blocking pattern
Ok(0)
}
@@ -658,7 +653,7 @@ impl<'sock> UdsStreamScheme<'sock> {
match option {
libc::SO_DOMAIN => write_value(&AF_UNIX.to_le_bytes()),
libc::SO_PEERCRED => {
let (_, remote_rc) = self.get_connected_peer(id, MsgFlags::default())?;
let (_, remote_rc) = self.get_connected_peer(id)?;
let remote = remote_rc.borrow();
write_value(unsafe {
slice::from_raw_parts(
@@ -696,7 +691,7 @@ impl<'sock> UdsStreamScheme<'sock> {
let (bytes_written, remote_id) = {
let name = self.get_socket(id)?.borrow().path.clone();
let (remote_id, remote_rc) = self.get_connected_peer(id, msg_flags)?;
let (remote_id, remote_rc) = self.get_connected_peer(id)?;
let mut socket = remote_rc.borrow_mut();
let connection = socket.require_connected_connection(msg_flags)?;
let (pid, uid, gid) = get_uid_gid_from_pid(self.proc_creds_capability, ctx.pid)?;
@@ -825,7 +820,7 @@ impl<'sock> UdsStreamScheme<'sock> {
}
fn handle_get_peer_name(&self, id: usize, payload: &mut [u8]) -> Result<usize> {
let (_, socket_rc) = self.get_connected_peer(id, MsgFlags::default())?;
let (_, socket_rc) = self.get_connected_peer(id)?;
let socket_borrow = socket_rc.borrow();
match socket_borrow.path.as_ref() {
Some(path_string) => Self::fpath_inner(path_string, payload),
@@ -850,7 +845,7 @@ impl<'sock> UdsStreamScheme<'sock> {
let new = listener_socket.accept(new_id, client_id, ctx)?;
let mut client_socket = client_rc.borrow_mut();
client_socket.establish(new_id)?;
client_socket.establish(new_id, listener_socket.primary_id)?;
(new_id, new)
};
@@ -877,15 +872,21 @@ impl<'sock> UdsStreamScheme<'sock> {
);
return Err(Error::new(EINVAL));
}
// Try to accept a waiting connection
let Some(client_id) = socket.awaiting.pop_front() else {
if flags & O_NONBLOCK == O_NONBLOCK {
return Err(Error::new(EAGAIN));
} else {
return Ok(Some(OpenResult::WouldBlock));
}
};
Ok(self.accept_connection(socket, client_id, ctx)?)
loop {
// Try to accept a waiting connection
let Some(client_id) = socket.awaiting.pop_front() else {
if flags & O_NONBLOCK == O_NONBLOCK {
return Err(Error::new(EAGAIN));
} else {
return Err(Error::new(EWOULDBLOCK));
}
};
return match self.accept_connection(socket, client_id, ctx) {
Ok(conn) => Ok(conn),
Err(Error { errno: EAGAIN }) => continue,
Err(e) => Err(e),
};
}
}
// Transition a Bound or Unbound socket to the Listening state.
@@ -932,10 +933,10 @@ impl<'sock> UdsStreamScheme<'sock> {
fn handle_connect_socketpair(&mut self, id: usize, ctx: &CallerCtx) -> Result<OpenResult> {
let new_id = self.next_id;
let flags = self.get_socket(id)?.borrow().flags;
let new = Socket::new(
let mut new = Socket::new(
new_id,
None,
State::Connecting,
State::Unbound,
HashSet::new(),
flags,
None,
@@ -952,7 +953,7 @@ impl<'sock> UdsStreamScheme<'sock> {
);
return Err(Error::new(EPIPE));
}
socket.connect_socketpair(new_id);
socket.connect_unchecked(&mut new);
}
// smoltcp sends writeable whenever a listener gets a
@@ -1237,7 +1238,7 @@ impl<'sock> SchemeSync for UdsStreamScheme<'sock> {
_flags: u32,
ctx: &CallerCtx,
) -> Result<usize> {
let (receiver_id, _) = self.get_connected_peer(id, MsgFlags::default())?;
let (receiver_id, _) = self.get_connected_peer(id)?;
self.write_inner(receiver_id, buf, ctx)
}
@@ -1277,7 +1278,7 @@ impl<'sock> SchemeSync for UdsStreamScheme<'sock> {
fn on_sendfd(&mut self, sendfd_request: &SendFdRequest) -> Result<usize> {
let id = sendfd_request.id();
let (receiver_id, _) = self.get_connected_peer(id, MsgFlags::default())?;
let (receiver_id, _) = self.get_connected_peer(id)?;
self.sendfd_inner(receiver_id, sendfd_request)
}