diff --git a/problem07/Cargo.lock b/problem07/Cargo.lock new file mode 100644 index 0000000..c8d1f98 --- /dev/null +++ b/problem07/Cargo.lock @@ -0,0 +1,241 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "anyhow" +version = "1.0.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" + +[[package]] +name = "bitflags" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "either" +version = "1.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "itertools" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b4baf93f58d4425749ca49a51c50ebab072c5df6994d08fed93541c331481dc" +dependencies = [ + "either", +] + +[[package]] +name = "libc" +version = "0.2.186" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + +[[package]] +name = "mio" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "problem07" +version = "0.1.0" +dependencies = [ + "anyhow", + "itertools", + "tokio", +] + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "syn" +version = "2.0.118" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tokio" +version = "1.52.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +dependencies = [ + "bytes", + "libc", + "mio", + "parking_lot", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] diff --git a/problem07/Cargo.toml b/problem07/Cargo.toml new file mode 100644 index 0000000..1dc04f5 --- /dev/null +++ b/problem07/Cargo.toml @@ -0,0 +1,9 @@ +[package] +name = "problem07" +version = "0.1.0" +edition = "2024" + +[dependencies] +anyhow = "1" +tokio = { version = "1", features = ["full"] } +itertools = "0" diff --git a/problem07/src/main.rs b/problem07/src/main.rs new file mode 100644 index 0000000..5736809 --- /dev/null +++ b/problem07/src/main.rs @@ -0,0 +1,448 @@ +use std::{ + collections::BTreeMap, + net::SocketAddr, + time::{Duration, Instant}, +}; + +use anyhow::{Result, anyhow, bail, ensure}; +use itertools::Itertools; +use tokio::net::UdpSocket; + +const RETRANSMISSION_DURATION: Duration = Duration::from_secs(3); +const SESSION_EXPIRY_DURATION: Duration = Duration::from_secs(60); + +#[tokio::main] +async fn main() -> Result<()> { + let socket = tokio::net::UdpSocket::bind("0.0.0.0:10000").await?; + + let mut read_buffer = vec![0; 1000]; + + let mut states = BTreeMap::new(); + + let mut timer = tokio::time::interval(RETRANSMISSION_DURATION); + + loop { + tokio::select! { + r = socket.recv_from(&mut read_buffer) => { + let (length, addr) = r?; + + let message = match String::from_utf8(read_buffer[..length].to_vec()) { + Ok(s) => s, + Err(e) => { + dbg!(e); + continue; + } + }; + + let message_type = match MessageType::parse(&message) { + Ok(message) => message, + Err(e) => { + dbg!(e); + continue; + } + }; + + let session = message_type.session(); + + let state = states.entry((addr, session)).or_insert(State::Closed); + + match message_type { + MessageType::Ack(session, length) => match state { + State::Closed => send_close(&socket, addr, session).await?, + + #[allow(clippy::nonminimal_bool)] + State::Open(open) => { + if !(length > open.highest_len_acked_so_far) { + continue; + } + if length > open.sent_so_far { + send_close(&socket, addr, session).await?; + *state = State::Closed; + states.remove(&(addr, session)); + continue; + } + if length < open.sent_so_far { + retransmit_payload_data(&socket, addr, open, length).await?; + } + if length > open.highest_len_acked_so_far { + open.highest_len_acked_so_far = length; + } + if open.highest_len_acked_so_far >= open.sent_so_far { + open.unacked_since = None; + } + } + }, + MessageType::Close(session) => { + send_close(&socket, addr, session).await?; + *state = State::Closed; + states.remove(&(addr, session)); + continue; + } + MessageType::Connect(session) => match state { + State::Closed => { + *state = State::Open(Open::new(session)); + let ack_message = send_ack(&socket, addr, session, 0).await?; + state.get_open()?.previous_ack = Some(ack_message); + } + State::Open(_) => { + let ack_message = send_ack(&socket, addr, session, 0).await?; + state.get_open()?.previous_ack = Some(ack_message); + } + }, + MessageType::Data(session, pos, data) => match state { + State::Closed => send_close(&socket, addr, session).await?, + State::Open(open) => { + let ack_message = if open.received_up_to == pos { + let unescaped = unescape(&data); + let length = unescaped.len() + open.received_up_to as usize; + + let ack_message = send_ack(&socket, addr, session, length).await?; + + open.buffer.push_str(&unescaped); + + while open.buffer.contains('\n') { + let line = open.reverse_next_line(); + let escaped = escape(&line); + send_data(&socket, addr, session, open.sent_so_far, escaped.clone()) + .await?; + open.payload_data.push_str(&unescape(escaped.as_str())); + open.sent_so_far += line.len() as u32; + open.unacked_since = Some(Instant::now()); + } + + open.received_up_to = length as u32; + + Some(ack_message) + } else { + match &open.previous_ack { + Some(ack) => { + send_previous_ack(&socket, addr, ack.clone()).await?; + } + None => { + let ack_message = send_ack(&socket, addr, session, 0).await?; + state.get_open()?.previous_ack = Some(ack_message); + } + } + None + }; + + if let Some(ack_message) = ack_message { + state.get_open()?.previous_ack = Some(ack_message); + } + } + }, + } + } + _ = timer.tick() => { + let now = Instant::now(); + let mut expired = vec![]; + for ((addr, _), state) in states.iter_mut() { + #[allow(clippy::collapsible_if)] + if let State::Open(open) = state { + if open.highest_len_acked_so_far < open.sent_so_far { + if let Some(since) = open.unacked_since { + if now.duration_since(since) >= SESSION_EXPIRY_DURATION { + send_close(&socket, *addr, open.session).await?; + expired.push(addr); + } else if now.duration_since(open.last_retransmit) >= RETRANSMISSION_DURATION { + let tail = open.payload_data + .chars() + .enumerate() + .filter(|(idx, _)| *idx >= open.highest_len_acked_so_far as usize) + .map(|(_, s)| s) + .collect::(); + send_data(&socket, *addr, open.session, open.highest_len_acked_so_far, tail).await?; + open.last_retransmit = now; + } + } + } + } + } + } + } + } +} + +async fn retransmit_payload_data( + socket: &UdpSocket, + addr: SocketAddr, + open: &Open, + length: u32, +) -> Result<()> { + let mut pos = None; + let mut data = String::new(); + for (idx, ch) in open + .payload_data + .chars() + .enumerate() + .filter(|(idx, _)| *idx >= length as usize) + { + if pos.is_none() { + pos = Some(idx) + } + data.push(ch); + } + + send_data(socket, addr, open.session, pos.unwrap() as u32, data).await?; + + Ok(()) +} + +async fn send_previous_ack( + socket: &UdpSocket, + addr: SocketAddr, + previous_ack: String, +) -> Result<()> { + socket.send_to(&previous_ack.into_bytes(), addr).await?; + Ok(()) +} + +async fn send_close(socket: &UdpSocket, addr: SocketAddr, session: u32) -> Result<()> { + socket + .send_to(&format!("/close/{session}/").into_bytes(), addr) + .await?; + Ok(()) +} + +async fn send_ack( + socket: &UdpSocket, + addr: SocketAddr, + session: u32, + length: usize, +) -> Result { + let message = format!("/ack/{session}/{length}/"); + socket.send_to(&message.clone().into_bytes(), addr).await?; + Ok(message) +} + +async fn send_data( + socket: &UdpSocket, + addr: SocketAddr, + session: u32, + mut pos: u32, + data: String, +) -> Result<()> { + let message = format!("/data/{session}/{pos}/{data}/"); + if message.len() < 1_000 { + socket.send_to(&message.into_bytes(), addr).await?; + } else { + let divisor = (message.len() as f64 / 1_000.).ceil() as usize; + let length = message.len() as usize / divisor - 100; // safety margin + let mut pointer = 0; + loop { + let chunk: String = data.chars().skip(pointer).take(length).collect(); + if chunk.is_empty() { + break; + } + + let message = format!("/data/{session}/{pos}/{chunk}/"); + socket.send_to(&message.into_bytes(), addr).await?; + + let unescaped_length = unescape(&chunk).len(); + + pointer += length; + pos += unescaped_length as u32; + } + } + Ok(()) +} + +#[derive(Debug)] +enum State { + Closed, + Open(Open), +} + +impl State { + fn get_open(&mut self) -> Result<&mut Open> { + match self { + State::Closed => Err(anyhow!("not open")), + State::Open(open) => Ok(open), + } + } +} + +#[derive(Debug)] +struct Open { + previous_ack: Option, + received_up_to: u32, + buffer: String, + sent_so_far: u32, + highest_len_acked_so_far: u32, + payload_data: String, + unacked_since: Option, + last_retransmit: Instant, + session: u32, +} + +impl Open { + fn new(session: u32) -> Self { + Self { + previous_ack: None, + received_up_to: 0, + buffer: String::new(), + sent_so_far: 0, + highest_len_acked_so_far: 0, + payload_data: String::new(), + unacked_since: None, + last_retransmit: Instant::now(), + session, + } + } + + fn reverse_next_line(&mut self) -> String { + // we checked outside whether this contains a newline + let idx = self.buffer.find('\n').unwrap(); + let mut line: String = self + .buffer + .drain(0..=idx) + .take_while(|c| *c != '\n') + .collect(); + line = line.chars().rev().collect(); + line.push('\n'); + line + } +} + +enum MessageType { + Ack(u32, u32), + Close(u32), + Connect(u32), + Data(u32, u32, String), +} + +impl MessageType { + #[allow(unstable_name_collisions)] + fn parse(s: &str) -> Result { + dbg!(s); + ensure!(s.starts_with('/')); + ensure!(s.ends_with('/')); + + ensure!(s.len() < 1_000); + + let mut split = s.split('/'); + split.next(); // omit first '/' which is now an empty string + + let message_type = split.next().ok_or(anyhow!("malformed message"))?; + + let result = match message_type { + "ack" => { + let session = + parse_num(split.next().ok_or(anyhow!("missing session identifier"))?)?; + + let length = parse_num(split.next().ok_or(anyhow!("ack missing length"))?)?; + ensure!(split.next() == Some(""), "ack packet missed trailing '/'"); + ensure!( + split.next().is_none(), + "extra fields in ack packet? {:?}", + split.next() + ); + + MessageType::Ack(session, length) + } + "close" => { + let session = parse_num(&parse_session_packet(&mut split)?)?; + MessageType::Close(session) + } + "connect" => { + let session = parse_num(&parse_session_packet(&mut split)?)?; + + MessageType::Connect(session) + } + "data" => { + let session = parse_num( + split + .next() + .ok_or_else(|| anyhow!("missing session identifier"))?, + )?; + + let position = parse_num( + split + .next() + .ok_or_else(|| anyhow!("missing position identifier"))?, + )?; + + ensure!( + split.next_back() == Some(""), + "data packet missed trailing '/'" + ); + + let data: Vec<_> = split.clone().collect(); + for i in 0..data.len() - 1 { + // all but ultimate segment must end with '\', otherwise there's inproper + // escaping + ensure!( + data[i].ends_with('\\'), + "extra fields in data part: {data:?}" + ); + } + + let data = split.intersperse("/").collect(); + + // let some = split.nth(1); // nth(1) because of clone() just earlier + // ensure!(some.is_none(), "extra fields in data packet? {some:?}",); + + MessageType::Data(session, position, data) + } + _ => bail!("unknown message type"), + }; + + Ok(result) + } + + fn session(&self) -> u32 { + match self { + MessageType::Ack(s, _) => *s, + MessageType::Close(s) => *s, + MessageType::Connect(s) => *s, + MessageType::Data(s, _, _) => *s, + } + } +} + +fn parse_num(s: &str) -> Result { + let n: u64 = s.parse()?; + ensure!(n < 2147483648); + Ok(n as u32) +} + +fn parse_session_packet(split: &mut std::str::Split<'_, char>) -> Result { + let session = split + .next() + .ok_or(anyhow!("missing session identifier"))? + .to_string(); + ensure!( + split.next() == Some(""), + "session packet missed trailing '/'" + ); + let some = split.next(); + ensure!(some.is_none(), "extra fields in session packet?: {some:?}",); + Ok(session) +} + +fn escape(s: &str) -> String { + s.replace('\\', "\\\\").replace('/', "\\/") +} + +fn unescape(s: &str) -> String { + s.replace("\\\\", "\\").replace("\\/", "/") +} + +#[cfg(test)] +mod tests { + use crate::MessageType; + + #[test] + fn parse_data_packet() { + let s = "/data/1234567/0/hey\\/huhu\\\\bar/"; + let parsed = MessageType::parse(s).unwrap(); + match parsed { + MessageType::Data(session, pos, data) => { + assert_eq!(session, 1234567); + assert_eq!(pos, 0); + assert_eq!(data, "hey\\/huhu\\\\bar"); + } + _ => panic!("wrong type parsed"), + } + } +}