add problem 07

2026-07-15 20:57:39 +02:00
parent 924964fcfb
commit e98de342ff
3 changed files with 698 additions and 0 deletions
@@ -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",
+]
@@ -0,0 +1,9 @@
+[package]
+name = "problem07"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+anyhow = "1"
+tokio = { version = "1", features = ["full"] }
+itertools = "0"
@@ -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::<String>();
+                                    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<String> {
+    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<String>,
+    received_up_to: u32,
+    buffer: String,
+    sent_so_far: u32,
+    highest_len_acked_so_far: u32,
+    payload_data: String,
+    unacked_since: Option<Instant>,
+    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<Self> {
+        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<u32> {
+    let n: u64 = s.parse()?;
+    ensure!(n < 2147483648);
+    Ok(n as u32)
+}
+
+fn parse_session_packet(split: &mut std::str::Split<'_, char>) -> Result<String> {
+    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"),
+        }
+    }
+}