diff --git a/problem09/Cargo.lock b/problem09/Cargo.lock index 77aa2e0..f8d8e47 100644 --- a/problem09/Cargo.lock +++ b/problem09/Cargo.lock @@ -26,26 +26,6 @@ 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" @@ -56,18 +36,6 @@ dependencies = [ "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" @@ -140,7 +108,6 @@ name = "problem09" version = "0.1.0" dependencies = [ "anyhow", - "rand", "serde", "serde_json", "tokio", @@ -164,29 +131,6 @@ 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" diff --git a/problem09/Cargo.toml b/problem09/Cargo.toml index 13b70e4..fc60773 100644 --- a/problem09/Cargo.toml +++ b/problem09/Cargo.toml @@ -8,4 +8,3 @@ 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 index 0d29ade..7d2a0d8 100644 --- a/problem09/src/main.rs +++ b/problem09/src/main.rs @@ -19,6 +19,7 @@ use anyhow::{Result, bail}; type Queue = BTreeMap>; static NEXT_CONN_ID: AtomicU64 = AtomicU64::new(0); +static NEXT_JOB_ID: AtomicU64 = AtomicU64::new(0); #[tokio::main] async fn main() -> Result<()> { @@ -68,7 +69,7 @@ async fn release_jobs_for_connection( } drop(map); if released { - tx.send(Notification::QueueExtended)?; + let _ = tx.send(Notification::QueueExtended); } Ok(()) } @@ -77,6 +78,10 @@ fn next_connection_id() -> ConnectionId { ConnectionId(NEXT_CONN_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed)) } +fn next_job_id() -> u64 { + NEXT_JOB_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed) +} + async fn handle_connection( mut socket: TcpStream, these_queues: Arc>, @@ -90,23 +95,27 @@ async fn handle_connection( loop { read_to_buffer(&mut reader, &mut read_buffer).await?; - let message = extract_next_message(&mut read_buffer)?; + let message = match extract_next_message(&mut read_buffer) { + Ok(message) => message, + Err(e) => { + let server_message = ServerMessage::error(e.to_string()); + write_message(&mut writer, &server_message).await?; + continue; + } + }; 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?; - + write_message(&mut writer, &server_message).await?; continue; } }; match client_message { ClientMessage::Put { queue, job, pri } => { - handle_put_request(&mut writer, queue, job, pri, &these_queues, tx).await?; + 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?; @@ -115,20 +124,43 @@ async fn handle_connection( handle_delete_request(&mut writer, id, &these_queues).await?; } ClientMessage::Abort { id } => { - handle_abort_request(&mut writer, id, conn_id).await?; + handle_abort_request(&mut writer, &these_queues, id, conn_id, &tx).await?; } } - - todo!() } } async fn handle_abort_request( writer: &mut (impl AsyncWriteExt + Unpin), + queues: &Arc>, id: u64, conn_id: ConnectionId, + tx: &Sender, ) -> Result<()> { - todo!() + let mut ok = false; + + let server_message = { + let mut btree_map = queues.lock().await; + let found = btree_map.values_mut().flatten().find(|j| j.id == id); + + match found { + None => ServerMessage::no_job(), + Some(job) if job.owner == Some(conn_id) => { + job.owner = None; + ok = true; + ServerMessage::ok() + } + Some(_) => ServerMessage::error("job not being worked on by this client".to_string()), + } + }; + + if ok { + let _ = tx.send(Notification::QueueExtended); + } + + write_message(writer, &server_message).await?; + + Ok(()) } async fn handle_delete_request( @@ -163,8 +195,7 @@ async fn handle_delete_request( None => ServerMessage::no_job(), }; - let reply = serde_json::to_string(&server_message)?; - writer.write_all(&reply.into_bytes()).await?; + write_message(writer, &server_message).await?; Ok(()) } @@ -175,8 +206,14 @@ async fn handle_put_request( job: Value, pri: u64, the_queues: &Arc>, - tx: Sender, + tx: &Sender, ) -> Result<()> { + if queue.is_empty() { + let server_message = ServerMessage::error("queue must not be empty".to_string()); + write_message(writer, &server_message).await?; + return Ok(()); + } + let job = Job::new(job, pri); let id = job.id; @@ -188,10 +225,9 @@ async fn handle_put_request( .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?; + write_message(writer, &server_message).await?; - tx.send(Notification::QueueExtended)?; + let _ = tx.send(Notification::QueueExtended); Ok(()) } @@ -204,21 +240,26 @@ async fn handle_get_request( tx: &Sender, conn_id: ConnectionId, ) -> Result<()> { + if queues.is_empty() { + let server_message = ServerMessage::error("queues must not be empty".to_string()); + write_message(writer, &server_message).await?; + return Ok(()); + } + 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)) => { + Some((queue, 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?; + write_message(writer, &server_message).await?; return Ok(()); } @@ -231,17 +272,16 @@ async fn handle_get_request( }; 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?; + write_message(writer, &server_message).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 +fn find_highest_pri_job_across_all_queues<'a>( + btree_map: &'a mut tokio::sync::MutexGuard<'_, Queue>, + queues: &'_ [String], +) -> Option<(String, &'a mut Job)> { + btree_map .iter_mut() .filter(|(s, _)| queues.contains(*s)) .filter_map(|(s, vec)| { @@ -250,9 +290,7 @@ fn find_highest_pri_job_across_all_queues( .max_by_key(|j| j.pri) .map(|j| (s.clone(), j)) }) - .max_by_key(|(_, j)| j.pri); - - best.map(|(s, j)| (s, j.clone())) + .max_by_key(|(_, j)| j.pri) } fn extract_next_message(read_buffer: &mut Vec) -> Result { @@ -279,6 +317,16 @@ async fn read_to_buffer( Ok(()) } +async fn write_message( + writer: &mut (impl AsyncWriteExt + Unpin), + msg: &ServerMessage, +) -> Result<()> { + let mut reply = serde_json::to_string(msg)?; + reply.push('\n'); + writer.write_all(&reply.into_bytes()).await?; + Ok(()) +} + #[derive(Clone, Debug, PartialEq)] struct Job { job: Value, @@ -289,7 +337,7 @@ struct Job { impl Job { fn new(job: Value, pri: u64) -> Self { - let id = rand::random::(); + let id = next_job_id(); let owner = None; Self { job,