finish 09

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"

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"

problem09/src/main.rs

@@ -19,6 +19,7 @@ use anyhow::{Result, bail};
 type Queue = BTreeMap<String, Vec<Job>>;
 
 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<Mutex<Queue>>,
@@ -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());
-
+                write_message(&mut writer, &server_message).await?;
-                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?;
+                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<Mutex<Queue>>,
     id: u64,
     conn_id: ConnectionId,
+    tx: &Sender<Notification>,
 ) -> 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)?;
+    write_message(writer, &server_message).await?;
-    writer.write_all(&reply.into_bytes()).await?;
 
     Ok(())
 }
@@ -175,8 +206,14 @@ async fn handle_put_request(
     job: Value,
     pri: u64,
     the_queues: &Arc<Mutex<Queue>>,
-    tx: Sender<Notification>,
+    tx: &Sender<Notification>,
 ) -> 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)?;
+    write_message(writer, &server_message).await?;
-    writer.write_all(&reply.into_bytes()).await?;
 
-    tx.send(Notification::QueueExtended)?;
+    let _ = tx.send(Notification::QueueExtended);
 
     Ok(())
 }
@@ -204,21 +240,26 @@ async fn handle_get_request(
     tx: &Sender<Notification>,
     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)?;
+                    write_message(writer, &server_message).await?;
-                    writer.write_all(&reply.into_bytes()).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)?;
+    write_message(writer, &server_message).await?;
-    writer.write_all(&reply.into_bytes()).await?;
 
     Ok(())
 }
 
-fn find_highest_pri_job_across_all_queues(
+fn find_highest_pri_job_across_all_queues<'a>(
-    btree_map: &mut tokio::sync::MutexGuard<'_, BTreeMap<String, Vec<Job>>>,
+    btree_map: &'a mut tokio::sync::MutexGuard<'_, Queue>,
-    queues: &[String],
+    queues: &'_ [String],
-) -> Option<(String, Job)> {
+) -> Option<(String, &'a mut Job)> {
-    let best = btree_map
+    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);
+        .max_by_key(|(_, j)| j.pri)
-
-    best.map(|(s, j)| (s, j.clone()))
 }
 
 fn extract_next_message(read_buffer: &mut Vec<u8>) -> Result<String> {
@@ -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::<u64>();
+        let id = next_job_id();
         let owner = None;
         Self {
             job,