From 3c07eff7b3bce06ac83baae7508bc6f1d875f150 Mon Sep 17 00:00:00 2001 From: timeshifter Date: Thu, 16 Jul 2026 22:42:16 +0200 Subject: [PATCH] add problem 09 (WIP) --- problem09/Cargo.lock | 344 +++++++++++++++++++++++++++++++++++ problem09/Cargo.toml | 11 ++ problem09/src/main.rs | 410 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 765 insertions(+) create mode 100644 problem09/Cargo.lock create mode 100644 problem09/Cargo.toml create mode 100644 problem09/src/main.rs diff --git a/problem09/Cargo.lock b/problem09/Cargo.lock new file mode 100644 index 0000000..77aa2e0 --- /dev/null +++ b/problem09/Cargo.lock @@ -0,0 +1,344 @@ +# 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.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" + +[[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 = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core", +] + +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + +[[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 = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "rand_core", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[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 = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[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 = "problem09" +version = "0.1.0" +dependencies = [ + "anyhow", + "rand", + "serde", + "serde_json", + "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 = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[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 = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[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.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tokio" +version = "1.52.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "317fafbbe3f02fc663dad00ea6186197de963cd4190e86a26d8d0fae095539af" +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", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/problem09/Cargo.toml b/problem09/Cargo.toml new file mode 100644 index 0000000..13b70e4 --- /dev/null +++ b/problem09/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "problem09" +version = "0.1.0" +edition = "2024" + +[dependencies] +anyhow = "1" +tokio = {version = "1", features = ["full"]} +serde = {version = "1", features = ["derive"]} +serde_json = "1" +rand = "0.10" diff --git a/problem09/src/main.rs b/problem09/src/main.rs new file mode 100644 index 0000000..0d29ade --- /dev/null +++ b/problem09/src/main.rs @@ -0,0 +1,410 @@ +use std::{ + collections::BTreeMap, + sync::{Arc, atomic::AtomicU64}, +}; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + sync::{ + Mutex, + broadcast::{self, Sender}, + }, +}; + +use anyhow::{Result, bail}; + +type Queue = BTreeMap>; + +static NEXT_CONN_ID: AtomicU64 = AtomicU64::new(0); + +#[tokio::main] +async fn main() -> Result<()> { + let listener = TcpListener::bind("0.0.0.0:10000").await?; + + let queues = Arc::new(Mutex::new(Queue::new())); + let (tx, _) = broadcast::channel(100_000); + + loop { + let (socket, _) = listener.accept().await?; + + let conn_id = next_connection_id(); + let these_queues = queues.clone(); + let this_tx = tx.clone(); + + tokio::spawn(async move { + match handle_connection(socket, these_queues.clone(), this_tx.clone(), conn_id).await { + Ok(()) => (), + Err(e) => { + dbg!(e); + } + } + match release_jobs_for_connection(these_queues, conn_id, this_tx).await { + Ok(()) => (), + Err(e) => { + dbg!(e); + } + }; + }); + } +} + +async fn release_jobs_for_connection( + queues: Arc>, + conn_id: ConnectionId, + tx: Sender, +) -> Result<()> { + let mut released = false; + let mut map = queues.lock().await; + for jobs in map.values_mut() { + for job in jobs.iter_mut() { + if job.owner == Some(conn_id) { + job.owner = None; + released = true; + } + } + } + drop(map); + if released { + tx.send(Notification::QueueExtended)?; + } + Ok(()) +} + +fn next_connection_id() -> ConnectionId { + ConnectionId(NEXT_CONN_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed)) +} + +async fn handle_connection( + mut socket: TcpStream, + these_queues: Arc>, + tx: Sender, + conn_id: ConnectionId, +) -> Result<()> { + let (mut reader, mut writer) = socket.split(); + + let mut read_buffer = Vec::with_capacity(128_000); + + loop { + read_to_buffer(&mut reader, &mut read_buffer).await?; + + let message = extract_next_message(&mut read_buffer)?; + + let client_message: ClientMessage = match serde_json::from_str(&message) { + Ok(parsed) => parsed, + Err(e) => { + let server_message = ServerMessage::error(e.to_string()); + + let reply = serde_json::to_string(&server_message).expect("we are screwed"); + writer.write_all(&reply.into_bytes()).await?; + + continue; + } + }; + + match client_message { + ClientMessage::Put { queue, job, pri } => { + handle_put_request(&mut writer, queue, job, pri, &these_queues, tx).await?; + } + ClientMessage::Get { queues, wait } => { + handle_get_request(&mut writer, queues, wait, &these_queues, &tx, conn_id).await?; + } + ClientMessage::Delete { id } => { + handle_delete_request(&mut writer, id, &these_queues).await?; + } + ClientMessage::Abort { id } => { + handle_abort_request(&mut writer, id, conn_id).await?; + } + } + + todo!() + } +} + +async fn handle_abort_request( + writer: &mut (impl AsyncWriteExt + Unpin), + id: u64, + conn_id: ConnectionId, +) -> Result<()> { + todo!() +} + +async fn handle_delete_request( + writer: &mut (impl AsyncWriteExt + Unpin), + id: u64, + these_queues: &Mutex, +) -> Result<()> { + let mut btree_map = these_queues.lock().await; + let mut queue_to_delete_in = None; + + 'outer: for (queue, jobs) in btree_map.iter_mut() { + for job in jobs { + if job.id == id { + queue_to_delete_in = Some(queue.to_string()); + break 'outer; + } + } + } + + let server_message = match queue_to_delete_in { + Some(queue) => { + let vec = btree_map.get_mut(&queue).unwrap(); + let pos = vec.iter().position(|j| j.id == id).unwrap(); + vec.remove(pos); + + if vec.is_empty() { + btree_map.remove(&queue); + } + + ServerMessage::ok() + } + None => ServerMessage::no_job(), + }; + + let reply = serde_json::to_string(&server_message)?; + writer.write_all(&reply.into_bytes()).await?; + + Ok(()) +} + +async fn handle_put_request( + writer: &mut (impl AsyncWriteExt + Unpin), + queue: String, + job: Value, + pri: u64, + the_queues: &Arc>, + tx: Sender, +) -> Result<()> { + let job = Job::new(job, pri); + let id = job.id; + + the_queues + .lock() + .await + .entry(queue) + .or_insert_with(Vec::new) + .push(job); + + let server_message = ServerMessage::ok_with_id(id); + let reply = serde_json::to_string(&server_message)?; + writer.write_all(&reply.into_bytes()).await?; + + tx.send(Notification::QueueExtended)?; + + Ok(()) +} + +async fn handle_get_request( + writer: &mut (impl AsyncWriteExt + Unpin), + queues: Vec, + wait: bool, + the_queues: &Arc>, + tx: &Sender, + conn_id: ConnectionId, +) -> Result<()> { + let (queue, job) = loop { + let mut btree_map = the_queues.lock().await; + + let best = find_highest_pri_job_across_all_queues(&mut btree_map, &queues); + + match best { + Some((queue, mut job)) => { + job.owner = Some(conn_id); + break (queue, job.clone()); + } + None => { + if !wait { + let server_message = ServerMessage::no_job(); + let reply = serde_json::to_string(&server_message)?; + writer.write_all(&reply.into_bytes()).await?; + + return Ok(()); + } + let mut rx = tx.subscribe(); + drop(btree_map); + rx.recv().await?; + continue; + } + } + }; + + let server_message = ServerMessage::ok_with_job(job.id, job.job.clone(), job.pri, queue); + let reply = serde_json::to_string(&server_message)?; + writer.write_all(&reply.into_bytes()).await?; + + Ok(()) +} + +fn find_highest_pri_job_across_all_queues( + btree_map: &mut tokio::sync::MutexGuard<'_, BTreeMap>>, + queues: &[String], +) -> Option<(String, Job)> { + let best = btree_map + .iter_mut() + .filter(|(s, _)| queues.contains(*s)) + .filter_map(|(s, vec)| { + vec.iter_mut() + .filter(|j| j.owner.is_none()) + .max_by_key(|j| j.pri) + .map(|j| (s.clone(), j)) + }) + .max_by_key(|(_, j)| j.pri); + + best.map(|(s, j)| (s, j.clone())) +} + +fn extract_next_message(read_buffer: &mut Vec) -> Result { + let next_newline = read_buffer + .iter() + .position(|c| *c == b'\n') + .expect("buffer was filled with newline"); + + let chunk: Vec<_> = read_buffer.drain(0..=next_newline).collect(); + let result = String::from_utf8(chunk)?; + Ok(result) +} + +async fn read_to_buffer( + reader: &mut (impl AsyncReadExt + Unpin), + read_buffer: &mut Vec, +) -> Result<(), anyhow::Error> { + while !read_buffer.contains(&b'\n') { + let read_amount = reader.read_buf(read_buffer).await?; + if read_amount == 0 { + bail!("client disconnected") + } + } + Ok(()) +} + +#[derive(Clone, Debug, PartialEq)] +struct Job { + job: Value, + pri: u64, + id: u64, + owner: Option, +} + +impl Job { + fn new(job: Value, pri: u64) -> Self { + let id = rand::random::(); + let owner = None; + Self { + job, + pri, + id, + owner, + } + } +} + +#[derive(Serialize, Debug)] +struct ServerMessage { + status: Status, + #[serde(skip_serializing_if = "Option::is_none")] + id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + job: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pri: Option, + #[serde(skip_serializing_if = "Option::is_none")] + queue: Option, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, +} + +impl ServerMessage { + fn ok_with_id(id: u64) -> Self { + Self { + status: Status::Ok, + id: Some(id), + job: None, + pri: None, + queue: None, + error: None, + } + } + + fn ok() -> Self { + Self { + status: Status::Ok, + id: None, + job: None, + pri: None, + queue: None, + error: None, + } + } + + fn ok_with_job(id: u64, job: Value, pri: u64, queue: String) -> Self { + Self { + status: Status::Ok, + id: Some(id), + job: Some(job), + pri: Some(pri), + queue: Some(queue), + error: None, + } + } + + fn no_job() -> Self { + Self { + status: Status::NoJob, + id: None, + job: None, + pri: None, + queue: None, + error: None, + } + } + + fn error(error: String) -> Self { + Self { + status: Status::Error, + id: None, + job: None, + pri: None, + queue: None, + error: Some(error), + } + } +} + +#[derive(Deserialize, Debug)] +#[serde(rename_all = "kebab-case", tag = "request", deny_unknown_fields)] +enum ClientMessage { + Put { + queue: String, + job: serde_json::Value, + pri: u64, + }, + Get { + queues: Vec, + #[serde(default)] + wait: bool, + }, + Delete { + id: u64, + }, + Abort { + id: u64, + }, +} + +#[derive(Serialize, Debug)] +#[serde(rename_all = "kebab-case")] +enum Status { + Ok, + Error, + NoJob, +} + +#[derive(Debug, Clone)] +enum Notification { + QueueExtended, +} + +#[derive(Debug, Clone, Copy, PartialEq)] +struct ConnectionId(u64);