github/wasm-pong
1
0
Fork 0
mirror of https://github.com/thilo-behnke/wasm-pong.git synced 2026-07-12 11:09:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

severe isses with concurrency - why is the ws read task starving while the write thread can still loop?

Browse Source
This commit is contained in:
Thilo Behnke
2022-06-06 17:30:53 +02:00
parent 89b2466aba
commit bcea6c6f0d
2 changed files with 30 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
server/src/http.rs
View File

@@ -156,33 +156,35 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe
}
println!("!!!! Exit websocket receiver !!!!")
});
// let websocket_session_write_copy = websocket_session.clone();
// tokio::spawn(async move {
// println!("Ready to read messages from kafka: {:?}", websocket_session_write_copy);
// loop {
// println!("Reading messages from kafka.");
// let messages = event_reader.read_from_session();
// if let Err(_) = messages {
// eprintln!("Failed to read messages from kafka for session: {:?}", websocket_session_write_copy);
// continue;
// }
// // println!("Read messages for websocket_session {:?} from consumer: {:?}", websocket_session_write_copy, messages);
// let messages = messages.unwrap();
// if messages.len() == 0 {
// println!("No new messages from kafka.");
// continue;
// }
// println!("{} new messages from kafka.", messages.len());
// let json = serde_json::to_string(&messages).unwrap();
// let message = Message::from(json);
// println!("Sending kafka messages through websocket.");
// let send_res = websocket_writer.send(message).await;
// if let Err(e) = send_res {
// eprintln!("Failed to send message to websocket for session {:?}: {:?}", websocket_session_write_copy, e)
// }
// sleep(Duration::from_millis(2000));
// }
// });
let websocket_session_write_copy = websocket_session.clone();
tokio::spawn(async move {
println!("Ready to read messages from kafka: {:?}", websocket_session_write_copy);
loop {
println!("Reading messages from kafka.");
let messages = event_reader.read_from_session();
if let Err(_) = messages {
eprintln!("Failed to read messages from kafka for session: {:?}", websocket_session_write_copy);
continue;
}
// println!("Read messages for websocket_session {:?} from consumer: {:?}", websocket_session_write_copy, messages);
let messages = messages.unwrap();
if messages.len() == 0 {
println!("No new messages from kafka.");
continue;
}
println!("{} new messages from kafka.", messages.len());
let json = serde_json::to_string(&messages).unwrap();
let message = Message::from(json);
println!("Sending kafka messages through websocket.");
let send_res = websocket_writer.send(message).await;
if let Err(e) = send_res {
eprintln!("Failed to send message to websocket for session {:?}: {:?}", websocket_session_write_copy, e)
}
// Avoid starvation of read thread (?)
// TODO: How to avoid this? This is very bad for performance.
sleep(Duration::from_millis(10)).await;
}
});
Ok(())
}

2
www/index.js
View File

@@ -102,7 +102,7 @@ const tick = () => {
const latestMoveEvents = Object.entries(moveEventsByObj)
.map(([_, moveEvents]) => moveEvents[moveEvents.length - 1]);
objects = latestMoveEvents.map(({msg}) => msg);
// sendEvents(getInputEvents())
sendEvents(getInputEvents())
}
render(objects);
}