fix(ipc): empty ipc msg crash (#1351)

* fix(ipc): recover from corrupted channel state

* style(fmt): rustfmt
This commit is contained in:
Aram Drevekenin 2022-04-27 10:44:14 +02:00 committed by GitHub
parent 90da35f4e6
commit 7ba49658f7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 116 additions and 85 deletions

View file

@ -58,11 +58,14 @@ fn assert_socket(name: &str) -> bool {
Ok(stream) => { Ok(stream) => {
let mut sender = IpcSenderWithContext::new(stream); let mut sender = IpcSenderWithContext::new(stream);
sender.send(ClientToServerMsg::ConnStatus); sender.send(ClientToServerMsg::ConnStatus);
let mut receiver: IpcReceiverWithContext<ServerToClientMsg> = sender.get_receiver(); let mut receiver: IpcReceiverWithContext<ServerToClientMsg> = sender.get_receiver();
let (instruction, _) = receiver.recv(); match receiver.recv() {
Some((instruction, _)) => {
matches!(instruction, ServerToClientMsg::Connected) matches!(instruction, ServerToClientMsg::Connected)
} }
None => false,
}
}
Err(e) if e.kind() == io::ErrorKind::ConnectionRefused => { Err(e) if e.kind() == io::ErrorKind::ConnectionRefused => {
drop(fs::remove_file(path)); drop(fs::remove_file(path));
false false

View file

@ -270,7 +270,8 @@ pub fn start_client(
let os_input = os_input.clone(); let os_input = os_input.clone();
let mut should_break = false; let mut should_break = false;
move || loop { move || loop {
let (instruction, err_ctx) = os_input.recv_from_server(); match os_input.recv_from_server() {
Some((instruction, err_ctx)) => {
err_ctx.update_thread_ctx(); err_ctx.update_thread_ctx();
if let ServerToClientMsg::Exit(_) = instruction { if let ServerToClientMsg::Exit(_) = instruction {
should_break = true; should_break = true;
@ -280,6 +281,14 @@ pub fn start_client(
break; break;
} }
} }
None => {
send_client_instructions
.send(ClientInstruction::UnblockInputThread)
.unwrap();
log::error!("Received empty message from server");
}
}
}
}) })
.unwrap(); .unwrap();

View file

@ -93,7 +93,7 @@ pub trait ClientOsApi: Send + Sync {
fn send_to_server(&self, msg: ClientToServerMsg); fn send_to_server(&self, msg: ClientToServerMsg);
/// Receives a message on client-side IPC channel /// Receives a message on client-side IPC channel
// This should be called from the client-side router thread only. // This should be called from the client-side router thread only.
fn recv_from_server(&self) -> (ServerToClientMsg, ErrorContext); fn recv_from_server(&self) -> Option<(ServerToClientMsg, ErrorContext)>;
fn handle_signals(&self, sigwinch_cb: Box<dyn Fn()>, quit_cb: Box<dyn Fn()>); fn handle_signals(&self, sigwinch_cb: Box<dyn Fn()>, quit_cb: Box<dyn Fn()>);
/// Establish a connection with the server socket. /// Establish a connection with the server socket.
fn connect_to_server(&self, path: &Path); fn connect_to_server(&self, path: &Path);
@ -144,7 +144,7 @@ impl ClientOsApi for ClientOsInputOutput {
.unwrap() .unwrap()
.send(msg); .send(msg);
} }
fn recv_from_server(&self) -> (ServerToClientMsg, ErrorContext) { fn recv_from_server(&self) -> Option<(ServerToClientMsg, ErrorContext)> {
self.receive_instructions_from_server self.receive_instructions_from_server
.lock() .lock()
.unwrap() .unwrap()

View file

@ -151,7 +151,7 @@ impl ClientOsApi for FakeClientOsApi {
command_is_executing.unblock_input_thread(); command_is_executing.unblock_input_thread();
} }
} }
fn recv_from_server(&self) -> (ServerToClientMsg, ErrorContext) { fn recv_from_server(&self) -> Option<(ServerToClientMsg, ErrorContext)> {
unimplemented!() unimplemented!()
} }
fn handle_signals(&self, _sigwinch_cb: Box<dyn Fn()>, _quit_cb: Box<dyn Fn()>) { fn handle_signals(&self, _sigwinch_cb: Box<dyn Fn()>, _quit_cb: Box<dyn Fn()>) {

View file

@ -397,7 +397,8 @@ pub(crate) fn route_thread_main(
client_id: ClientId, client_id: ClientId,
) { ) {
loop { loop {
let (instruction, err_ctx) = receiver.recv(); match receiver.recv() {
Some((instruction, err_ctx)) => {
err_ctx.update_thread_ctx(); err_ctx.update_thread_ctx();
let rlocked_sessions = session_data.read().unwrap(); let rlocked_sessions = session_data.read().unwrap();
@ -405,10 +406,18 @@ pub(crate) fn route_thread_main(
ClientToServerMsg::Action(action) => { ClientToServerMsg::Action(action) => {
if let Some(rlocked_sessions) = rlocked_sessions.as_ref() { if let Some(rlocked_sessions) = rlocked_sessions.as_ref() {
if let Action::SwitchToMode(input_mode) = action { if let Action::SwitchToMode(input_mode) = action {
os_input os_input.send_to_client(
.send_to_client(client_id, ServerToClientMsg::SwitchToMode(input_mode)); client_id,
ServerToClientMsg::SwitchToMode(input_mode),
);
} }
if route_action(action, rlocked_sessions, &*os_input, &to_server, client_id) { if route_action(
action,
rlocked_sessions,
&*os_input,
&to_server,
client_id,
) {
break; break;
} }
} }
@ -435,7 +444,9 @@ pub(crate) fn route_thread_main(
.as_ref() .as_ref()
.unwrap() .unwrap()
.senders .senders
.send_to_screen(ScreenInstruction::TerminalPixelDimensions(pixel_dimensions)) .send_to_screen(ScreenInstruction::TerminalPixelDimensions(
pixel_dimensions,
))
.unwrap(); .unwrap();
} }
ClientToServerMsg::NewClient( ClientToServerMsg::NewClient(
@ -475,4 +486,9 @@ pub(crate) fn route_thread_main(
} }
} }
} }
None => {
log::error!("Received empty message from client");
}
}
}
} }

View file

@ -184,8 +184,11 @@ where
} }
/// Receives an event, along with the current [`ErrorContext`], on this [`IpcReceiverWithContext`]'s socket. /// Receives an event, along with the current [`ErrorContext`], on this [`IpcReceiverWithContext`]'s socket.
pub fn recv(&mut self) -> (T, ErrorContext) { pub fn recv(&mut self) -> Option<(T, ErrorContext)> {
bincode::deserialize_from(&mut self.receiver).unwrap() match bincode::deserialize_from(&mut self.receiver) {
Ok(msg) => Some(msg),
Err(_) => None,
}
} }
/// Returns an [`IpcSenderWithContext`] with the same socket as this receiver. /// Returns an [`IpcSenderWithContext`] with the same socket as this receiver.