From fbcbc8dfa1f02e2aed3ba13aa05793082c4a30f5 Mon Sep 17 00:00:00 2001 From: Sean Cross Date: Tue, 28 Jul 2020 18:12:52 +0800 Subject: [PATCH] kernel: fix deadlock in threaded messages Due to an ordering issue, it was possible for thread message responses to become deadlocked, wherein a message would get stored in the shared table in between checking the table value and locking the network connection. This change makes a second check of the table after the network connection is locked, in order to prevent this condition. Signed-off-by: Sean Cross --- kernel/src/arch/hosted.rs | 348 ++++++++++++-------- kernel/src/test.rs | 633 ++++++++++++++++++++----------------- xous-rs/src/arch/hosted.rs | 57 ++-- 3 files changed, 595 insertions(+), 443 deletions(-) diff --git a/kernel/src/arch/hosted.rs b/kernel/src/arch/hosted.rs index 83fa5d9a8..da6806276 100644 --- a/kernel/src/arch/hosted.rs +++ b/kernel/src/arch/hosted.rs @@ -4,14 +4,12 @@ pub mod process; pub mod syscall; use std::cell::RefCell; +use std::convert::TryInto; use std::env; use std::io::Read; -use std::mem::size_of; use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener, TcpStream, ToSocketAddrs}; use std::sync::mpsc::{channel, Receiver, Sender}; -use std::thread::spawn; use std::thread_local; -use std::time::Duration; use crate::arch::process::Process; use crate::services::SystemServices; @@ -61,112 +59,164 @@ pub fn set_send_addr(send_addr: Sender) { /// Each client gets its own connection and its own thread, which is handled here. fn handle_connection( - mut conn: TcpStream, + conn: TcpStream, pid: PID, chn: Sender, should_exit: std::sync::Arc, ) { - loop { - let mut pkt = [0usize; 9]; - let mut incoming_word = [0u8; size_of::()]; - conn.set_read_timeout(Some(Duration::from_millis(1000))) - .unwrap(); - for word in pkt.iter_mut() { - loop { - if should_exit.load(core::sync::atomic::Ordering::Relaxed) { - chn.send(ThreadMessage::SysCall( - pid, - 1, - xous::SysCall::TerminateProcess, - )) - .unwrap(); - } - if let Err(e) = conn.read_exact(&mut incoming_word) { - // If the connection has gone away, send a `TerminateProcess` message to the main - // and then exit this thread. - if e.kind() == std::io::ErrorKind::WouldBlock - || e.kind() == std::io::ErrorKind::TimedOut - { - continue; - } else { - // println!( - // "KERNEL({}): Client disconnected: {} ({:?}). Shutting down virtual process.", - // pid, e, e - // ); - chn.send(ThreadMessage::SysCall( - pid, - 1, - xous::SysCall::TerminateProcess, - )) - .unwrap(); - return; - } - } - break; - } - *word = usize::from_le_bytes(incoming_word); - } + enum ServerMessage { + Exit, + ServerPacket([usize; 9]), + ServerPacketWithData([usize; 9], Vec), + } - let thread_id = pkt[0]; - let call = xous::SysCall::from_args( - pkt[1], pkt[2], pkt[3], pkt[4], pkt[5], pkt[6], pkt[7], pkt[8], - ); - - match call { - Err(e) => eprintln!("KERNEL({}): Received invalid syscall: {:?}", pid, e), - Ok(mut call) => { - if let SysCall::SendMessage(ref _cid, ref mut envelope) = call { - match envelope { - xous::Message::MutableBorrow(msg) - | xous::Message::Borrow(msg) - | xous::Message::Move(msg) => { - let mut tmp_data = vec![0; msg.buf.len()]; - conn.read_exact(&mut tmp_data) - .map_err(|_e| { - // println!("KERNEL({}): Read Error {}", pid, _e); - chn.send(ThreadMessage::SysCall( - pid, - thread_id, - xous::SysCall::TerminateProcess, - )) - .unwrap(); - }) - .unwrap(); - // Update the address pointer. This will get turned back into a - // usable pointer by casting it back into a &[T] on the other - // side. This is just a pointer to the start of data - // as well as the index into the data it points at. The lengths - // should still be equal once we reconstitute the data in the - // other process. - // ::debug_here::debug_here!(); - let sliced_data = tmp_data.into_boxed_slice(); - assert_eq!( - sliced_data.len(), - msg.buf.len(), - "deconstructed data {} != message buf length {}", - sliced_data.len(), - msg.buf.len() - ); - msg.buf.addr = match MemoryAddress::new(Box::into_raw(sliced_data) - as *mut usize - as usize) - { - Some(a) => a, - _ => unreachable!(), - }; - } - xous::Message::Scalar(_) => (), - } - } + fn conn_thread(mut conn: TcpStream, sender: Sender) { + loop { + let mut raw_data = [0u8; 9 * std::mem::size_of::()]; + if let Err(_e) = conn.read_exact(&mut raw_data) { // println!( - // "Received packet: {:08x} {} {} {} {} {} {} {}: {:?}", - // pkt[0], pkt[1], pkt[2], pkt[3], pkt[4], pkt[5], pkt[6], pkt[7], call + // "KERNEL(?): Client disconnected: {} ({:?}). Shutting down virtual process.", + // _e, _e // ); - chn.send(ThreadMessage::SysCall(pid, thread_id, call)) - .expect("couldn't make syscall"); + sender.send(ServerMessage::Exit).ok(); + return; + } + + let mut packet_data = [0usize; 9]; + for (bytes, word) in raw_data + .chunks_exact(std::mem::size_of::()) + .zip(packet_data.iter_mut()) + { + *word = usize::from_le_bytes(bytes.try_into().unwrap()); + } + + if packet_data[1] == 16 + && (packet_data[3] == 1 || packet_data[3] == 2 || packet_data[3] == 3) + { + let mut v = vec![0; packet_data[6]]; + if conn.read_exact(&mut v).is_err() { + sender.send(ServerMessage::Exit).ok(); + return; + } + sender + .send(ServerMessage::ServerPacketWithData(packet_data, v)) + .unwrap(); + } else { + sender + .send(ServerMessage::ServerPacket(packet_data)) + .unwrap(); } } } + + let (sender, receiver) = channel(); + let conn_sender = sender.clone(); + std::thread::Builder::new() + .name(format!("PID {}: client connection thread", pid)) + .spawn(move || { + conn_thread(conn, conn_sender); + }) + .unwrap(); + + std::thread::Builder::new() + .name(format!("PID {}: client should_exit thread", pid)) + .spawn(move || loop { + if should_exit.load(core::sync::atomic::Ordering::Relaxed) { + // eprintln!("KERNEL: should_exit == 1"); + sender.send(ServerMessage::Exit).ok(); + return; + } + std::thread::park_timeout(std::time::Duration::from_secs(1)); + }) + .unwrap(); + + for msg in receiver { + match msg { + ServerMessage::Exit => break, + ServerMessage::ServerPacket(pkt) => { + let thread_id = pkt[0]; + let call = xous::SysCall::from_args( + pkt[1], pkt[2], pkt[3], pkt[4], pkt[5], pkt[6], pkt[7], pkt[8], + ); + match call { + Err(e) => { + eprintln!("KERNEL({}): Received invalid syscall: {:?}", pid, e); + eprintln!( + "Raw packet: {:08x} {} {} {} {} {} {} {}", + pkt[0], pkt[1], pkt[2], pkt[3], pkt[4], pkt[5], pkt[6], pkt[7] + ); + } + Ok(call) => chn + .send(ThreadMessage::SysCall(pid, thread_id, call)) + .expect("couldn't make syscall"), + } + } + ServerMessage::ServerPacketWithData(pkt, data) => { + let thread_id = pkt[0]; + let call = xous::SysCall::from_args( + pkt[1], pkt[2], pkt[3], pkt[4], pkt[5], pkt[6], pkt[7], pkt[8], + ); + match call { + Err(e) => { + eprintln!("KERNEL({}): Received invalid syscall: {:?}", pid, e); + eprintln!( + "Raw packet: {:08x} {} {} {} {} {} {} {}", + pkt[0], pkt[1], pkt[2], pkt[3], pkt[4], pkt[5], pkt[6], pkt[7] + ); + } + Ok(mut call) => { + // eprintln!( + // "Received packet: {:08x} {} {} {} {} {} {} {}: {:?}", + // pkt[0], pkt[1], pkt[2], pkt[3], pkt[4], pkt[5], pkt[6], pkt[7], call + // ); + if let SysCall::SendMessage(ref _cid, ref mut envelope) = call { + match envelope { + xous::Message::MutableBorrow(msg) + | xous::Message::Borrow(msg) + | xous::Message::Move(msg) => { + // Update the address pointer. This will get turned back into a + // usable pointer by casting it back into a &[T] on the other + // side. This is just a pointer to the start of data + // as well as the index into the data it points at. The lengths + // should still be equal once we reconstitute the data in the + // other process. + // ::debug_here::debug_here!(); + let sliced_data = data.into_boxed_slice(); + assert_eq!( + sliced_data.len(), + msg.buf.len(), + "deconstructed data {} != message buf length {}", + sliced_data.len(), + msg.buf.len() + ); + msg.buf.addr = + match MemoryAddress::new(Box::into_raw(sliced_data) + as *mut usize + as usize) + { + Some(a) => a, + _ => unreachable!(), + }; + } + xous::Message::Scalar(_) => (), + } + } else { + panic!("unsupported message type"); + } + chn.send(ThreadMessage::SysCall(pid, thread_id, call)) + .expect("couldn't make syscall"); + } + } + } + } + } + // eprintln!("KERNEL({}): Finished the thread so sending TerminateProcess", pid); + chn.send(ThreadMessage::SysCall( + pid, + 1, + xous::SysCall::TerminateProcess, + )) + .unwrap(); } fn listen_thread( @@ -215,7 +265,10 @@ fn listen_thread( // println!("KERNEL({}): New client connected from {}", new_pid, _addr); let conn_copy = conn.try_clone().expect("couldn't duplicate connection"); let should_exit = should_exit.clone(); - let jh = spawn(move || handle_connection(conn, new_pid, thr_chn, should_exit)); + let jh = std::thread::Builder::new() + .name(format!("kernel PID {} listener", new_pid)) + .spawn(move || handle_connection(conn, new_pid, thr_chn, should_exit)) + .expect("couldn't spawn listen thread"); clients.push((jh, conn_copy)); false } @@ -233,12 +286,6 @@ fn listen_thread( } } - // let pid1 = listener.accept().unwrap(); - // if accept_new_connection(pid1.0, &chn, &backchannel, &mut clients, &should_exit) { - // exit_server(should_exit, clients); - // return; - // } - // Use `listener` in a nonblocking setup so that we can exit when doing tests enum ClientMessage { NewConnection(TcpStream), @@ -247,44 +294,72 @@ fn listen_thread( let (sender, receiver) = channel(); let tcp_sender = sender.clone(); let exit_sender = sender; - spawn(move || loop { - match listener.accept() { - Ok((conn, _addr)) => { - tcp_sender.send(ClientMessage::NewConnection(conn)).unwrap(); + + let (shutdown_listener, shutdown_listener_receiver) = channel(); + + // `listener.accept()` has no way to break, so we must put it in nonblocking mode + listener.set_nonblocking(true).unwrap(); + + std::thread::Builder::new() + .name("kernel accept thread".to_owned()) + .spawn(move || loop { + match listener.accept() { + Ok((conn, _addr)) => { + conn.set_nonblocking(false).unwrap(); + tcp_sender.send(ClientMessage::NewConnection(conn)).unwrap(); + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + match shutdown_listener_receiver.recv_timeout(std::time::Duration::from_millis(500)) { + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => continue, + Ok(()) | Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + return; + }, + } + } + Err(e) => { + // Windows generates this error -- WSACancelBlockingCall -- when a + // connection is shut down while `accept()` is running. This should + // only happen when the system is shutting down, so ignore it. + if cfg!(windows) { + if let Some(10004) = e.raw_os_error() { + return; + } + } + eprintln!( + "error accepting connections: {} ({:?}) ({:?})", + e, + e, + e.kind() + ); + return; + } } - Err(e) => { - eprintln!("error accepting connections: {} ({:?})", e, e); - return; - } - } - }); + }) + .unwrap(); // Spawn a thread to listen for the `exit` command, and relay that // to the main thread. This prevents us from needing to poll, since // all messages are coalesced into a single channel. - spawn(move || match exit_channel.recv() { - Ok(ExitMessage::Exit) => { - exit_sender.send(ClientMessage::Exit).unwrap() - } - Err(std::sync::mpsc::RecvError) => { - eprintln!("error receiving exit command") - } - }); + std::thread::Builder::new() + .name("kernel exit listener".to_owned()) + .spawn(move || match exit_channel.recv() { + Ok(ExitMessage::Exit) => exit_sender.send(ClientMessage::Exit).unwrap(), + Err(std::sync::mpsc::RecvError) => eprintln!("error receiving exit command"), + }) + .unwrap(); for msg in receiver { match msg { ClientMessage::NewConnection(conn) => { if accept_new_connection(conn, &chn, &new_pid_channel, &mut clients, &should_exit) { - exit_server(should_exit, clients); - return; + break; } } - ClientMessage::Exit => { - exit_server(should_exit, clients); - return; - } + ClientMessage::Exit => break, } } + shutdown_listener.send(()).unwrap(); + exit_server(should_exit, clients); } /// The idle function is run when there are no directly-runnable processes @@ -313,7 +388,10 @@ pub fn idle() -> bool { let listen_thread_handle = SEND_ADDR.with(|sa| { let sa = sa.borrow_mut().take(); - spawn(move || listen_thread(listen_addr, sender, sa, new_pid_receiver, exit_receiver)) + std::thread::Builder::new() + .name("kernel network listener".to_owned()) + .spawn(move || listen_thread(listen_addr, sender, sa, new_pid_receiver, exit_receiver)) + .expect("couldn't spawn listen thread") }); while let Ok(msg) = receiver.recv() { @@ -396,10 +474,10 @@ pub fn idle() -> bool { } process.send(&response_vec).unwrap_or_else(|_e| { // If we're unable to send data to the process, assume it's dead and terminate it. - // println!( - // "KERNEL({}): Unable to send response to process: {:?} -- terminating", - // pid, _e - // ); + eprintln!( + "KERNEL({}): Unable to send response to process: {:?} -- terminating", + pid, _e + ); crate::syscall::handle(pid, thread_id, SysCall::TerminateProcess).ok(); }); crate::arch::process::set_current_pid(existing_pid); diff --git a/kernel/src/test.rs b/kernel/src/test.rs index 04f6217f1..a7105cf99 100644 --- a/kernel/src/test.rs +++ b/kernel/src/test.rs @@ -47,15 +47,17 @@ fn start_kernel(server_spec: &str) -> JoinHandle<()> { // Launch the main thread. We pass a `send_addr` channel so that the // server can notify us when it's ready to listen. - let main_thread = std::thread::spawn(move || { - let server_spec_server = server_addr; - crate::arch::set_pid1_key(pid1_key); - crate::arch::set_send_addr(send_addr); - crate::arch::set_listen_address(&server_spec_server); - kmain() - }); + let main_thread = std::thread::Builder::new() + .name("kernel main".to_owned()) + .spawn(move || { + let server_spec_server = server_addr; + crate::arch::set_pid1_key(pid1_key); + crate::arch::set_send_addr(send_addr); + crate::arch::set_listen_address(&server_spec_server); + kmain() + }) + .expect("couldn't start kernel thread"); let server_addr = recv_addr.recv().unwrap(); - println!("Got server address: {:?}", server_addr); xous::arch::set_xous_address(server_addr); // Connect to server. This first instance needs to make sure the kernel is listening. @@ -77,7 +79,7 @@ fn start_kernel(server_spec: &str) -> JoinHandle<()> { fn shutdown_kernel() { // Any process ought to be able to shut down the system currently. xous::wait_process( - xous::create_process(xous::ProcessArgs::new(|| { + xous::create_process(xous::ProcessArgs::new("shutdown", || { rsyscall(SysCall::Shutdown).expect("unable to shutdown server"); })) .expect("couldn't shut down the kernel"), @@ -124,39 +126,46 @@ fn send_scalar_message() { // and receive the message. Note that we need to communicate to the // "Client" what our server ID is. Normally this would be done via // an external nameserver. - let xous_server = xous::create_process(xous::ProcessArgs::new(move || { - let sid = xous::create_server(b"send_scalar_mesg").expect("couldn't create test server"); - server_addr_send.send(sid).unwrap(); - let envelope = xous::receive_message(sid).expect("couldn't receive messages"); - assert_eq!( - envelope.message, - xous::Message::Scalar(xous::ScalarMessage { - id: 1, - arg1: 2, - arg2: 3, - arg3: 4, - arg4: 5 - }) - ); - })) + let xous_server = xous::create_process(xous::ProcessArgs::new( + "send_scalar_message server", + move || { + let sid = + xous::create_server(b"send_scalar_mesg").expect("couldn't create test server"); + server_addr_send.send(sid).unwrap(); + let envelope = xous::receive_message(sid).expect("couldn't receive messages"); + assert_eq!( + envelope.message, + xous::Message::Scalar(xous::ScalarMessage { + id: 1, + arg1: 2, + arg2: 3, + arg3: 4, + arg4: 5 + }) + ); + }, + )) .expect("couldn't spawn server process"); // Spawn the client "process" and wait for the server address. - let xous_client = xous::create_process(xous::ProcessArgs::new(move || { - let sid = server_addr_recv.recv().unwrap(); - let conn = xous::connect(sid).expect("couldn't connect to server"); - xous::send_message( - conn, - xous::Message::Scalar(xous::ScalarMessage { - id: 1, - arg1: 2, - arg2: 3, - arg3: 4, - arg4: 5, - }), - ) - .expect("couldn't send message"); - })) + let xous_client = xous::create_process(xous::ProcessArgs::new( + "send_scalar_message client", + move || { + let sid = server_addr_recv.recv().unwrap(); + let conn = xous::connect(sid).expect("couldn't connect to server"); + xous::send_message( + conn, + xous::Message::Scalar(xous::ScalarMessage { + id: 1, + arg1: 2, + arg2: 3, + arg3: 4, + arg4: 5, + }), + ) + .expect("couldn't send message"); + }, + )) .expect("couldn't spawn client process"); // Wait for both processes to finish @@ -176,36 +185,47 @@ fn send_move_message() { let (server_addr_send, server_addr_recv) = channel(); - let xous_server = xous::create_process(xous::ProcessArgs::new(move || { - let sid = xous::create_server(b"send_move_messag").expect("couldn't create test server"); - server_addr_send.send(sid).unwrap(); - let envelope = xous::receive_message(sid).expect("couldn't receive messages"); - // println!("Received message from {}", envelope.sender); - let message = envelope.message; - if let xous::Message::Move(m) = message { - let buf = m.buf; - let bt = unsafe { - Box::from_raw(core::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len())) - }; - assert_eq!(*test_bytes, *bt); - // let s = String::from_utf8_lossy(&bt); - // println!("Got message: {:?} -> \"{}\"", bt, s); - } else { - panic!("unexpected message type"); - } + let xous_server = xous::create_process(xous::ProcessArgs::new( + "send_move_message server", + move || { + // println!("SERVER: Creating server..."); + let sid = + xous::create_server(b"send_move_messag").expect("couldn't create test server"); + // println!("SERVER: Sending server address of {:?} to client", sid); + server_addr_send.send(sid).unwrap(); + // println!("SERVER: Starting to receive messages..."); + let envelope = xous::receive_message(sid).expect("couldn't receive messages"); + // println!("SERVER: Received message from {}", envelope.sender); + let message = envelope.message; + if let xous::Message::Move(m) = message { + let buf = m.buf; + let bt = unsafe { + Box::from_raw(core::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len())) + }; + assert_eq!(*test_bytes, *bt); + // let s = String::from_utf8_lossy(&bt); + // println!("SERVER: Got message: {:?} -> \"{}\"", bt, s); + } else { + panic!("unexpected message type"); + } + }, + )) + .expect("couldn't start server"); - // println!("SERVER: Received message: {:?}", msg); - })).expect("couldn't start server"); - - let xous_client = xous::create_process(xous::ProcessArgs::new(move || { - // println!("CLIENT: Waiting for server address..."); - let sid = server_addr_recv.recv().unwrap(); - // println!("CLIENT: Connecting to server {:?}", sid); - let conn = xous::connect(sid).expect("couldn't connect to server"); - let msg = xous::carton::Carton::from_bytes(test_bytes); - xous::send_message(conn, xous::Message::Move(msg.into_message(0))) - .expect("couldn't send a message"); - })).expect("couldn't start client"); + let xous_client = xous::create_process(xous::ProcessArgs::new( + "send_move_message client", + move || { + // println!("CLIENT: Waiting for server address..."); + let sid = server_addr_recv.recv().unwrap(); + // println!("CLIENT: Connecting to server {:?}", sid); + let conn = xous::connect(sid).expect("couldn't connect to server"); + let msg = xous::carton::Carton::from_bytes(test_bytes); + xous::send_message(conn, xous::Message::Move(msg.into_message(0))) + .expect("couldn't send a message"); + // println!("CLIENT: Message sent"); + }, + )) + .expect("couldn't start client"); // Wait for both processes to finish crate::wait_process(xous_server).expect("couldn't join server process"); @@ -224,50 +244,59 @@ fn send_borrow_message() { let test_str = "Hello, world!"; let test_bytes = test_str.as_bytes(); - let xous_server = xous::create_process(xous::ProcessArgs::new(move || { - // println!("SERVER: Creating server..."); - let sid = xous::create_server(b"send_borrow_mesg").expect("couldn't create test server"); - server_addr_send.send(sid).unwrap(); - // println!("SERVER: Receiving message..."); - let envelope = xous::receive_message(sid).expect("couldn't receive messages"); - // println!("SERVER: Received message from {}", envelope.sender); - let message = envelope.message; - if let xous::Message::Borrow(m) = message { - let buf = m.buf; - let bt = unsafe { - Box::from_raw(core::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len())) - }; - assert_eq!(*test_bytes, *bt); - // let s = String::from_utf8_lossy(&bt); - // println!("SERVER: Got message: {:?} -> \"{}\"", bt, s); - xous::return_memory(envelope.sender, m.buf).unwrap(); - // println!("SERVER: Returned memory"); - } else { - panic!("unexpected message type"); - } - })).expect("couldn't start server"); + let xous_server = xous::create_process(xous::ProcessArgs::new( + "send_borrow_message server", + move || { + // println!("SERVER: Creating server..."); + let sid = + xous::create_server(b"send_borrow_mesg").expect("couldn't create test server"); + server_addr_send.send(sid).unwrap(); + // println!("SERVER: Receiving message..."); + let envelope = xous::receive_message(sid).expect("couldn't receive messages"); + // println!("SERVER: Received message from {}", envelope.sender); + let message = envelope.message; + if let xous::Message::Borrow(m) = message { + let buf = m.buf; + let bt = unsafe { + Box::from_raw(core::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len())) + }; + assert_eq!(*test_bytes, *bt); + // let s = String::from_utf8_lossy(&bt); + // println!("SERVER: Got message: {:?} -> \"{}\"", bt, s); + xous::return_memory(envelope.sender, m.buf).unwrap(); + // println!("SERVER: Returned memory"); + } else { + panic!("unexpected message type"); + } + }, + )) + .expect("couldn't start server"); - let xous_client = xous::create_process(xous::ProcessArgs::new(move || { - // Get the server address (out of band) so we know what to connect to - // println!("CLIENT: Waiting for server to start..."); - let sid = server_addr_recv.recv().unwrap(); + let xous_client = xous::create_process(xous::ProcessArgs::new( + "send_borrow_message client", + move || { + // Get the server address (out of band) so we know what to connect to + // println!("CLIENT: Waiting for server to start..."); + let sid = server_addr_recv.recv().unwrap(); - // Perform a connection to the server - // println!("CLIENT: Connecting to server..."); - let conn = xous::connect(sid).expect("couldn't connect to server"); + // Perform a connection to the server + // println!("CLIENT: Connecting to server..."); + let conn = xous::connect(sid).expect("couldn't connect to server"); - // Convert the message into a "Carton" that can be shipped as a message - // println!("CLIENT: Creating carton..."); - let carton = xous::carton::Carton::from_bytes(test_bytes); + // Convert the message into a "Carton" that can be shipped as a message + // println!("CLIENT: Creating carton..."); + let carton = xous::carton::Carton::from_bytes(test_bytes); - // Send the message to the server - // println!("CLIENT: Lending message..."); - carton - .lend(conn, 0) - .expect("couldn't lend message to server"); + // Send the message to the server + // println!("CLIENT: Lending message..."); + carton + .lend(conn, 0) + .expect("couldn't lend message to server"); - // println!("CLIENT: Done"); - })).expect("couldn't start client"); + // println!("CLIENT: Done"); + }, + )) + .expect("couldn't start client"); // Wait for both processes to finish crate::wait_process(xous_server).expect("couldn't join server process"); @@ -286,48 +315,57 @@ fn send_mutableborrow_message() { let test_str = "Hello, world!"; let test_bytes = test_str.as_bytes(); - let xous_server = xous::create_process(xous::ProcessArgs::new(move || { - let sid = xous::create_server(b"send_mutborrow_m").expect("couldn't create test server"); - server_addr_send.send(sid).unwrap(); - let envelope = xous::receive_message(sid).expect("couldn't receive messages"); - // println!("Received message from {}", envelope.sender); - let message = envelope.message; - if let xous::Message::MutableBorrow(m) = message { - let buf = m.buf; - let mut bt = unsafe { - Box::from_raw(core::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len())) - }; - for letter in bt.iter_mut() { + let xous_server = xous::create_process(xous::ProcessArgs::new( + "send_mutableborrow_message server", + move || { + let sid = + xous::create_server(b"send_mutborrow_m").expect("couldn't create test server"); + server_addr_send.send(sid).unwrap(); + let envelope = xous::receive_message(sid).expect("couldn't receive messages"); + // println!("Received message from {}", envelope.sender); + let message = envelope.message; + if let xous::Message::MutableBorrow(m) = message { + let buf = m.buf; + let mut bt = unsafe { + Box::from_raw(core::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len())) + }; + for letter in bt.iter_mut() { + *letter += 1; + } + xous::return_memory(envelope.sender, m.buf).unwrap(); + } else { + panic!("unexpected message type"); + } + }, + )) + .expect("couldn't start server"); + + let xous_client = xous::create_process(xous::ProcessArgs::new( + "send_mutableborrow_message client", + move || { + // Get the server address (out of band) so we know what to connect to + let sid = server_addr_recv.recv().unwrap(); + + // Perform a connection to the server + let conn = xous::connect(sid).expect("couldn't connect to server"); + + // Convert the message into a "Carton" that can be shipped as a message + let mut carton = xous::carton::Carton::from_bytes(&test_bytes); + let mut check_bytes = test_bytes.to_vec(); + for letter in check_bytes.iter_mut() { *letter += 1; } - xous::return_memory(envelope.sender, m.buf).unwrap(); - } else { - panic!("unexpected message type"); - } - })).expect("couldn't start server"); - let xous_client = xous::create_process(xous::ProcessArgs::new(move || { - // Get the server address (out of band) so we know what to connect to - let sid = server_addr_recv.recv().unwrap(); + // Send the message to the server + carton + .lend_mut(conn, 3) + .expect("couldn't mutably lend data"); - // Perform a connection to the server - let conn = xous::connect(sid).expect("couldn't connect to server"); - - // Convert the message into a "Carton" that can be shipped as a message - let mut carton = xous::carton::Carton::from_bytes(&test_bytes); - let mut check_bytes = test_bytes.to_vec(); - for letter in check_bytes.iter_mut() { - *letter += 1; - } - - // Send the message to the server - carton - .lend_mut(conn, 3) - .expect("couldn't mutably lend data"); - - let modified_bytes: &[u8] = carton.as_ref(); - assert_eq!(&check_bytes, &modified_bytes); - })).expect("couldn't start client"); + let modified_bytes: &[u8] = carton.as_ref(); + assert_eq!(&check_bytes, &modified_bytes); + }, + )) + .expect("couldn't start client"); // Wait for both processes to finish crate::wait_process(xous_server).expect("couldn't join server process"); @@ -348,52 +386,61 @@ fn send_mutableborrow_message_repeat() { let loops = 50; - let xous_server = xous::create_process(xous::ProcessArgs::new(move || { - let sid = xous::create_server(b"send_mutborrow_r").expect("couldn't create test server"); - server_addr_send.send(sid).unwrap(); + let xous_server = xous::create_process(xous::ProcessArgs::new( + "send_mutableborrow_message_repeat server", + move || { + let sid = + xous::create_server(b"send_mutborrow_r").expect("couldn't create test server"); + server_addr_send.send(sid).unwrap(); - for iteration in 0..loops { - let envelope = xous::receive_message(sid).expect("couldn't receive messages"); - let message = envelope.message; - if let xous::Message::MutableBorrow(m) = message { - let buf = m.buf; - let mut bt = unsafe { - Box::from_raw(core::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len())) - }; - for letter in bt.iter_mut() { + for iteration in 0..loops { + let envelope = xous::receive_message(sid).expect("couldn't receive messages"); + let message = envelope.message; + if let xous::Message::MutableBorrow(m) = message { + let buf = m.buf; + let mut bt = unsafe { + Box::from_raw(core::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len())) + }; + for letter in bt.iter_mut() { + *letter = (*letter).wrapping_add((iteration & 0xff) as u8); + } + xous::return_memory(envelope.sender, m.buf).unwrap(); + } else { + panic!("unexpected message type"); + } + } + }, + )) + .expect("couldn't start server"); + + let xous_client = xous::create_process(xous::ProcessArgs::new( + "send_mutableborrow_message_repeat client", + move || { + // Get the server address (out of band) so we know what to connect to + let sid = server_addr_recv.recv().unwrap(); + + // Perform a connection to the server + let conn = xous::connect(sid).expect("couldn't connect to server"); + + // Convert the message into a "Carton" that can be shipped as a message + for iteration in 0..loops { + let mut carton = xous::carton::Carton::from_bytes(&test_bytes); + let mut check_bytes = test_bytes.to_vec(); + for letter in check_bytes.iter_mut() { *letter = (*letter).wrapping_add((iteration & 0xff) as u8); } - xous::return_memory(envelope.sender, m.buf).unwrap(); - } else { - panic!("unexpected message type"); + + // Send the message to the server + carton + .lend_mut(conn, 3) + .expect("couldn't mutably lend data"); + + let modified_bytes: &[u8] = carton.as_ref(); + assert_eq!(&check_bytes, &modified_bytes); } - } - })).expect("couldn't start server"); - - let xous_client = xous::create_process(xous::ProcessArgs::new(move || { - // Get the server address (out of band) so we know what to connect to - let sid = server_addr_recv.recv().unwrap(); - - // Perform a connection to the server - let conn = xous::connect(sid).expect("couldn't connect to server"); - - // Convert the message into a "Carton" that can be shipped as a message - for iteration in 0..loops { - let mut carton = xous::carton::Carton::from_bytes(&test_bytes); - let mut check_bytes = test_bytes.to_vec(); - for letter in check_bytes.iter_mut() { - *letter = (*letter).wrapping_add((iteration & 0xff) as u8); - } - - // Send the message to the server - carton - .lend_mut(conn, 3) - .expect("couldn't mutably lend data"); - - let modified_bytes: &[u8] = carton.as_ref(); - assert_eq!(&check_bytes, &modified_bytes); - } - })).expect("couldn't start client"); + }, + )) + .expect("couldn't start client"); // Wait for both processes to finish crate::wait_process(xous_server).expect("couldn't join server process"); @@ -439,24 +486,27 @@ fn server_client_same_process() { // Start the kernel in its own thread let main_thread = start_kernel(SERVER_SPEC); - let internal_server = xous::create_process(xous::arch::ProcessArgs::new(|| { - let server = xous::create_server(b"s_c_same_process").expect("couldn't create server"); - let connection = xous::connect(server).expect("couldn't connect to our own server"); - let msg_contents = xous::ScalarMessage { - id: 1, - arg1: 2, - arg2: 3, - arg3: 4, - arg4: 5, - }; + let internal_server = xous::create_process(xous::arch::ProcessArgs::new( + "server_client_same_process process", + || { + let server = xous::create_server(b"s_c_same_process").expect("couldn't create server"); + let connection = xous::connect(server).expect("couldn't connect to our own server"); + let msg_contents = xous::ScalarMessage { + id: 1, + arg1: 2, + arg2: 3, + arg3: 4, + arg4: 5, + }; - xous::send_message(connection, xous::Message::Scalar(msg_contents)) - .expect("couldn't send message"); + xous::send_message(connection, xous::Message::Scalar(msg_contents)) + .expect("couldn't send message"); - let msg = xous::receive_message(server).expect("couldn't receive message"); + let msg = xous::receive_message(server).expect("couldn't receive message"); - assert_eq!(msg.message, xous::Message::Scalar(msg_contents)); - })) + assert_eq!(msg.message, xous::Message::Scalar(msg_contents)); + }, + )) .expect("couldn't start server"); xous::wait_process(internal_server).expect("couldn't join internal_server process"); @@ -474,39 +524,42 @@ fn multiple_contexts() { // Start the kernel in its own thread let main_thread = start_kernel(SERVER_SPEC); - let internal_server = xous::create_process(xous::ProcessArgs::new(move || { - let server = xous::create_server(b"multiple_context").expect("couldn't create server"); - let connection = xous::connect(server).expect("couldn't connect to our own server"); - let msg_contents = xous::ScalarMessage { - id: 1, - arg1: 2, - arg2: 3, - arg3: 4, - arg4: 5, - }; + let internal_server = xous::create_process(xous::ProcessArgs::new( + "multiple_contexts process", + move || { + let server = xous::create_server(b"multiple_context").expect("couldn't create server"); + let connection = xous::connect(server).expect("couldn't connect to our own server"); + let msg_contents = xous::ScalarMessage { + id: 1, + arg1: 2, + arg2: 3, + arg3: 4, + arg4: 5, + }; - let mut server_threads = vec![]; - for _ in 0..crate::arch::process::MAX_THREAD { - server_threads.push( - xous::create_thread(move || { - let msg = xous::receive_message(server).expect("couldn't receive message"); - assert_eq!(msg.message, xous::Message::Scalar(msg_contents)); - }) - .expect("couldn't spawn client thread"), - ); - } + let mut server_threads = vec![]; + for _ in 0..crate::arch::process::MAX_THREAD { + server_threads.push( + xous::create_thread(move || { + let msg = xous::receive_message(server).expect("couldn't receive message"); + assert_eq!(msg.message, xous::Message::Scalar(msg_contents)); + }) + .expect("couldn't spawn client thread"), + ); + } - for _ in &server_threads { - xous::send_message(connection, xous::Message::Scalar(msg_contents)) - .expect("couldn't send message"); - } - for server_thread in server_threads.into_iter() { - xous::wait_thread(server_thread).expect("couldn't wait for thread"); - } - })).expect("couldn't create internal server"); + for _ in &server_threads { + xous::send_message(connection, xous::Message::Scalar(msg_contents)) + .expect("couldn't send message"); + } + for server_thread in server_threads.into_iter() { + xous::wait_thread(server_thread).expect("couldn't wait for thread"); + } + }, + )) + .expect("couldn't create internal server"); - xous::wait_process(internal_server) - .expect("couldn't join internal_server process"); + xous::wait_process(internal_server).expect("couldn't join internal_server process"); // Any process ought to be able to shut down the system currently. shutdown_kernel(); @@ -514,63 +567,77 @@ fn multiple_contexts() { main_thread.join().expect("couldn't join kernel process"); } -// #[test] -// fn multiple_multiple_contexts() { -// for _ in 0..5 { -// multiple_contexts(); -// } -// } +#[test] +fn multiple_multiple_contexts() { + for _ in 0..5 { + multiple_contexts(); + } +} -// /// Test that a server can be restarted and the kernel doesn't crash -// #[test] -// fn process_restart_server() { -// let test_str = "Hello, world!"; -// let test_bytes = test_str.as_bytes(); +/// Test that a server can be restarted and the kernel doesn't crash +#[test] +fn process_restart_server() { + let test_str = "Hello, world!"; + let test_bytes = test_str.as_bytes(); -// let main_thread = start_kernel(SERVER_SPEC); + let main_thread = start_kernel(SERVER_SPEC); -// fn create_destroy_server(test_bytes: &'static [u8]) { -// let (server_addr_send, server_addr_recv) = channel(); + fn create_destroy_server(test_bytes: &'static [u8]) { + let (server_addr_send, server_addr_recv) = channel(); -// let xous_server = as_process(move || { -// let sid = -// xous::create_server(b"test_recreate_se").expect("couldn't create test server"); -// server_addr_send.send(sid).unwrap(); -// let thr = xous::create_thread(move || { -// let envelope = xous::receive_message(sid).expect("couldn't receive messages"); -// // println!("Received message from {}", envelope.sender); -// let message = envelope.message; -// if let xous::Message::Move(m) = message { -// let buf = m.buf; -// let bt = unsafe { -// Box::from_raw(core::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len())) -// }; -// assert_eq!(*test_bytes, *bt); -// // let s = String::from_utf8_lossy(&bt); -// // println!("Got message: {:?} -> \"{}\"", bt, s); -// } else { -// panic!("unexpected message type"); -// } -// }) -// .unwrap(); -// xous::wait_thread(thr).unwrap(); -// }); + let xous_server = xous::create_process(xous::ProcessArgs::new( + "process_restart_server server", + move || { + let sid = + xous::create_server(b"test_recreate_se").expect("couldn't create test server"); + server_addr_send.send(sid).unwrap(); + let thr = xous::create_thread(move || { + let envelope = xous::receive_message(sid).expect("couldn't receive messages"); + // println!("Received message from {}", envelope.sender); + let message = envelope.message; + if let xous::Message::Move(m) = message { + let buf = m.buf; + let bt = unsafe { + Box::from_raw(core::slice::from_raw_parts_mut( + buf.as_mut_ptr(), + buf.len(), + )) + }; + assert_eq!(*test_bytes, *bt); + // let s = String::from_utf8_lossy(&bt); + // println!("Got message: {:?} -> \"{}\"", bt, s); + } else { + panic!("unexpected message type"); + } + }) + .unwrap(); + xous::wait_thread(thr).unwrap(); + }, + )) + .expect("couldn't spawn server process"); -// // Wait for the server to start up -// let sid = server_addr_recv.recv().unwrap(); + // Wait for the server to start up + let xous_client = xous::create_process(xous::ProcessArgs::new( + "process_restart_server client", + move || { + let sid = server_addr_recv.recv().unwrap(); + let conn = xous::connect(sid).expect("couldn't connect to server"); + let msg = xous::carton::Carton::from_bytes(test_bytes); + xous::send_message(conn, xous::Message::Move(msg.into_message(0))) + .expect("couldn't send a message"); + }, + )) + .expect("couldn't start client process"); -// let conn = xous::connect(sid).expect("couldn't connect to server"); -// let msg = xous::carton::Carton::from_bytes(test_bytes); -// xous::send_message(conn, xous::Message::Move(msg.into_message(0))) -// .expect("couldn't send a message"); -// xous_server.join().expect("couldn't join server process"); -// } + xous::wait_process(xous_server).expect("couldn't join server process"); + xous::wait_process(xous_client).expect("couldn't join client process"); + } -// // create_destroy_server(test_bytes); -// create_destroy_server(test_bytes); + // create_destroy_server(test_bytes); + create_destroy_server(test_bytes); -// // Any process ought to be able to shut down the system currently. -// rsyscall(SysCall::Shutdown).expect("unable to shutdown server"); + // Any process ought to be able to shut down the system currently. + shutdown_kernel(); -// main_thread.join().expect("couldn't join kernel process"); -// } + main_thread.join().expect("couldn't join kernel process"); +} diff --git a/xous-rs/src/arch/hosted.rs b/xous-rs/src/arch/hosted.rs index e763cd525..1403e1c3a 100644 --- a/xous-rs/src/arch/hosted.rs +++ b/xous-rs/src/arch/hosted.rs @@ -19,14 +19,18 @@ pub struct ProcessInit { pub struct ProcessArgs { main: Option, + name: String, } impl ProcessArgs where F: FnOnce(), { - pub fn new(main: F) -> ProcessArgs { - ProcessArgs { main: Some(main) } + pub fn new(name: &str, main: F) -> ProcessArgs { + ProcessArgs { + main: Some(main), + name: name.to_owned(), + } } } @@ -125,20 +129,6 @@ fn xous_address() -> SocketAddr { .unwrap_or_else(default_xous_address) } -// pub fn xous_connect() { -// XOUS_SERVER_CONNECTION.with(|xsc| { -// if xsc.borrow().is_none() { -// NETWORK_CONNECT_ADDRESS.with(|nca| { -// let addr = nca.borrow().unwrap_or_else(default_xous_address); -// match xous_connect_impl(addr) { -// Ok(a) => *xsc.borrow_mut() = Some(a), -// Err(_) => panic!("couldn't connect to server"), -// } -// }); -// } -// }) -// } - pub fn create_thread_pre(_f: &F) -> core::result::Result where F: FnOnce() -> T, @@ -210,7 +200,9 @@ where Ok(()) } })?; - Ok(ProcessInit { key: PROCESS_KEY.with(|pk| *pk.borrow()) }) + Ok(ProcessInit { + key: PROCESS_KEY.with(|pk| *pk.borrow()), + }) } pub fn create_process_post( @@ -223,7 +215,9 @@ where { let server_address = xous_address(); + let f = args.main.take().unwrap(); let thread_main = std::thread::Builder::new() + .name(args.name) .spawn(move || { set_xous_address(server_address); THREAD_ID.with(|tid| *tid.borrow_mut() = 1); @@ -239,19 +233,21 @@ where } })?; - crate::create_thread(args.main.take().unwrap()) + crate::create_thread(f) }) - .map_err(|_| crate::Error::InternalError)?.join().unwrap().unwrap(); + .map_err(|_| crate::Error::InternalError)? + .join() + .unwrap() + .unwrap(); Ok(ProcessHandle(thread_main.0)) } pub fn wait_process(joiner: ProcessHandle) -> crate::SysCallResult { - joiner - .0 - .join() - .map(|_| Result::Ok) - .map_err(|_| crate::Error::InternalError) + joiner.0.join().map(|_| Result::Ok).map_err(|_x| { + // panic!("wait error: {:?}", x); + crate::Error::InternalError + }) } fn xous_connect_impl( @@ -351,12 +347,23 @@ fn _xous_syscall_result( let mut stream = match server_connection.recv.try_lock() { Ok(lk) => lk, Err(std::sync::TryLockError::WouldBlock) => { - std::thread::sleep(std::time::Duration::from_millis(1)); + std::thread::sleep(std::time::Duration::from_millis(10)); continue; } Err(e) => panic!("Receive error: {}", e), }; + // One more check, in case something came in while we waited for the receiver above. + { + let mut mailbox = server_connection.mailbox.lock().unwrap(); + if let Some(entry) = mailbox.remove(&thread_id) { + if Result::BlockedProcess != entry { + *ret = entry; + return; + } + } + } + // This thread_id doesn't exist in the mailbox, so read additional data. let mut pkt = [0usize; 8]; let mut raw_bytes = [0u8; size_of::() * 9];