diff --git a/benches/storage_bench.rs b/benches/storage_bench.rs index b778739..5badc91 100644 --- a/benches/storage_bench.rs +++ b/benches/storage_bench.rs @@ -6,6 +6,7 @@ use futures::AsyncReadExt; use queueber::protocol; use queueber::protocol::queue; use queueber::storage::{RetriedStorage, Storage}; +// rand no longer needed; retries removed use std::net::{SocketAddr, TcpListener}; use std::sync::OnceLock; use std::sync::mpsc::sync_channel; @@ -26,7 +27,7 @@ fn bench_add_messages(c: &mut Criterion) { // Keep dir alive with storage by returning it as part of the state (dir, storage) }, - |(_dir, storage)| { + |(_dir, storage)| -> Result<(), queueber::errors::Error> { // Build the item once and reuse it; only IDs change per insert let mut msg = CapnpBuilder::new_default(); let mut item = msg.init_root::(); @@ -36,10 +37,11 @@ fn bench_add_messages(c: &mut Criterion) { for i in 0..num_items { let id = format!("id_{}", i); - storage - .add_available_item((id.as_bytes(), item_reader.reborrow())) - .expect("add_available_item"); + track_and_ignore_busy_error( + storage.add_available_item((id.as_bytes(), item_reader.reborrow())), + )?; } + Ok(()) }, BatchSize::SmallInput, ); @@ -85,13 +87,11 @@ fn bench_remove_messages(c: &mut Criterion) { (dir, storage, lease, ids) }, - |(_dir, storage, lease, ids)| { + |(_dir, storage, lease, ids)| -> Result<(), queueber::errors::Error> { for id in ids { - let removed = storage - .remove_in_progress_item(&id, &lease) - .expect("remove_in_progress_item"); - assert!(removed, "expected removal for id"); + track_and_ignore_busy_error(storage.remove_in_progress_item(&id, &lease))?; } + Ok(()) }, BatchSize::SmallInput, ); @@ -124,8 +124,9 @@ fn bench_poll_messages_storage(c: &mut Criterion) { } (dir, storage) }, - |(_dir, storage)| { - let _ = storage.get_next_available_entries(num_items).expect("poll"); + |(_dir, storage)| -> Result<(), queueber::errors::Error> { + let _ = track_and_ignore_busy_error(storage.get_next_available_entries(num_items))?; + Ok(()) }, BatchSize::SmallInput, ); @@ -283,7 +284,8 @@ fn bench_e2e_add_poll_remove(c: &mut Criterion) { item.set_contents(b"hello"); item.set_visibility_timeout_secs(0); } - let _ = request.send().promise.await.unwrap(); + let _ = + track_capnp_and_ignore_busy(request.send().promise.await, "rpc_add_batch"); }); }, |_| { @@ -293,8 +295,8 @@ fn bench_e2e_add_poll_remove(c: &mut Criterion) { req.set_lease_validity_secs(30); req.set_num_items(num_items); req.set_timeout_secs(0); - let reply = request.send().promise.await.unwrap(); - let _resp = reply.get().unwrap().get_resp().unwrap(); + let _ = + track_capnp_and_ignore_busy(request.send().promise.await, "rpc_poll_batch"); }); }, BatchSize::SmallInput, @@ -315,13 +317,20 @@ fn bench_e2e_add_poll_remove(c: &mut Criterion) { item.set_contents(b"hello"); item.set_visibility_timeout_secs(0); } - let _ = add.send().promise.await.unwrap(); + let _ = track_capnp_and_ignore_busy( + add.send().promise.await, + "rpc_remove_setup_add", + ); + let mut poll = queue_client.poll_request(); let mut preq = poll.get().init_req(); preq.set_lease_validity_secs(30); preq.set_num_items(num_items); preq.set_timeout_secs(0); - let reply = poll.send().promise.await.unwrap(); + let reply = match poll.send().promise.await { + Ok(r) => r, + Err(_) => return (vec![], vec![]), + }; let resp = reply.get().unwrap().get_resp().unwrap(); let lease = resp.get_lease().unwrap().to_vec(); let items = resp.get_items().unwrap(); @@ -339,7 +348,10 @@ fn bench_e2e_add_poll_remove(c: &mut Criterion) { let mut r = req.get().init_req(); r.set_id(&id); r.set_lease(&lease); - let _ = req.send().promise.await.unwrap(); + let _ = track_capnp_and_ignore_busy( + req.send().promise.await, + "rpc_remove_each", + ); } }); }, @@ -394,29 +406,48 @@ fn bench_e2e_stress_like(c: &mut Criterion) { req.set_lease_validity_secs(30); req.set_num_items(batch_size); req.set_timeout_secs(1); - let reply = request.send().promise.await.unwrap(); + let reply = match request.send().promise.await { + Ok(r) => r, + Err(_) => continue, + }; let resp = reply.get().unwrap().get_resp().unwrap(); let items = resp.get_items().unwrap(); let lease = resp.get_lease().unwrap(); + let items_len = items.len(); + let lease_vec = lease.to_vec(); + let lease = &lease_vec; if lease.len() == 16 { let mut lease_arr = [0u8; 16]; lease_arr.copy_from_slice(lease); current_lease = Some(lease_arr); } - if items.is_empty() { + if items_len == 0 { continue; } - let promises = items.iter().map(|i| { - let mut request = queue_client.remove_request(); - let mut r = request.get().init_req(); - r.set_id(i.get_id().unwrap()); - r.set_lease(lease); - request.send().promise - }); - let _ = futures::future::join_all(promises).await; - removed_count.fetch_add(items.len(), atomic::Ordering::Relaxed); + // Remove up to batch_size items by issuing batch_size remove calls. + // We don't have the ids from the fast path above, so poll again immediately with num_items=batch_size + let mut preq = queue_client.poll_request(); + let mut pr = preq.get().init_req(); + pr.set_lease_validity_secs(30); + pr.set_num_items(batch_size); + pr.set_timeout_secs(0); + if let Ok(reply) = preq.send().promise.await { + let resp = reply.get().unwrap().get_resp().unwrap(); + let items = resp.get_items().unwrap(); + let lease2 = resp.get_lease().unwrap(); + let promises = items.iter().map(|i| { + let mut request = queue_client.remove_request(); + let mut r = request.get().init_req(); + r.set_id(i.get_id().unwrap()); + r.set_lease(lease2); + request.send().promise + }); + let _ = futures::future::join_all(promises).await; + removed_count + .fetch_add(items.len(), atomic::Ordering::Relaxed); + } // Occasionally extend the current lease to exercise extend path if last_extend.elapsed() > std::time::Duration::from_secs(2) { @@ -425,7 +456,10 @@ fn bench_e2e_stress_like(c: &mut Criterion) { let mut xr = xreq.get().init_req(); xr.set_lease(&lease_arr); xr.set_lease_validity_secs(30); - let _ = xreq.send().promise.await; + let _ = track_capnp_and_ignore_busy( + xreq.send().promise.await, + "rpc_stress_extend", + ); } last_extend = std::time::Instant::now(); } @@ -473,7 +507,10 @@ fn bench_e2e_stress_like(c: &mut Criterion) { item.set_contents(b"p"); item.set_visibility_timeout_secs(0); } - let _ = request.send().promise.await.unwrap(); + let _ = track_capnp_and_ignore_busy( + request.send().promise.await, + "rpc_stress_add", + ); } }); }); @@ -495,3 +532,38 @@ criterion_group!( bench_e2e_stress_like ); criterion_main!(benches); + +// retry_capnp_busy removed; ignoring Busy errors inline instead + +fn track_and_ignore_busy_error( + res: Result, +) -> Result, queueber::errors::Error> { + match res { + Ok(v) => Ok(Some(v)), + Err(e) => { + if let queueber::errors::Error::Rocksdb { ref source, .. } = e + && source.kind() == rocksdb::ErrorKind::Busy + { + eprintln!("busy (ignored)"); + Ok(None) + } else { + Err(e) + } + } + } +} + +fn track_capnp_and_ignore_busy(res: Result, label: &str) -> Option { + match res { + Ok(v) => Some(v), + Err(e) => { + let msg = e.to_string(); + if msg.contains("Busy") || msg.contains("busy") || msg.contains("resource busy") { + eprintln!("{}: busy (ignored)", label); + None + } else { + panic!("{}", e); + } + } + } +} diff --git a/src/bin/client/main.rs b/src/bin/client/main.rs index f599b91..68d8830 100644 --- a/src/bin/client/main.rs +++ b/src/bin/client/main.rs @@ -145,16 +145,38 @@ async fn main() -> Result<()> { item.set_contents(contents.as_bytes()); item.set_visibility_timeout_secs(visibility_timeout_secs); - let reply = request.send().promise.await?; - let ids = reply.get()?.get_resp()?.get_ids()?; - - println!( - "received {:?} ids: {:?}", - ids.len(), - ids.iter() - .map(|id| -> Result { Ok(Uuid::from_slice(id?)?) }) - .collect::, _>>()? - ); + match request.send().promise.await { + Ok(reply) => { + let ids = reply.get()?.get_resp()?.get_ids()?; + let id_strs: Vec = (0..ids.len()) + .map(|i| { + let bytes = ids + .get(i) + .map_err(|e| capnp::Error::failed(e.to_string()))?; + Ok::( + Uuid::from_slice(bytes) + .map(|u| u.to_string()) + .unwrap_or_else(|_| format!("{:?}", bytes)), + ) + }) + .collect::>()?; + println!("received {:?} ids: {:?}", ids.len(), id_strs); + } + Err(e) => { + if is_capnp_busy_error(&e) { + tracing::warn!( + operation = "add", + total_reqs = 1, + busy_reqs = 1, + percent = 100.0, + "resource busy: ignoring" + ); + return Ok::<(), Box>(()); + } else { + return Err::<(), Box>(Box::new(e)); + } + } + } Ok::<(), Box>(()) }) .await? @@ -172,32 +194,48 @@ async fn main() -> Result<()> { req.set_num_items(num_items); req.set_timeout_secs(timeout_secs); - let reply = request.send().promise.await?; - let resp = reply.get()?.get_resp()?; - let lease = resp.get_lease()?; - let items = resp.get_items()?; - - println!( - "lease: {}", - Uuid::from_slice(lease) - .map(|u| u.to_string()) - .unwrap_or_else(|_| format!("{:?}", lease)) - ); - if items.is_empty() { - println!("no items available"); - } else { - for i in 0..items.len() { - let item = items.get(i); - let id = item.get_id()?; - let contents = item.get_contents()?; + match request.send().promise.await { + Ok(reply) => { + let resp = reply.get()?.get_resp()?; + let lease = resp.get_lease()?; + let items = resp.get_items()?; println!( - "item {}: id={}, contents=", - i, - Uuid::from_slice(id) + "lease: {}", + Uuid::from_slice(lease) .map(|u| u.to_string()) - .unwrap_or_else(|_| format!("{:?}", id)), + .unwrap_or_else(|_| format!("{:?}", lease)) ); - println!("{}", String::from_utf8_lossy(contents)); + if items.is_empty() { + println!("no items available"); + } else { + for i in 0..items.len() { + let item = items.get(i); + let id = item.get_id()?; + let contents = item.get_contents()?; + println!( + "item {}: id={}, contents=", + i, + Uuid::from_slice(id) + .map(|u| u.to_string()) + .unwrap_or_else(|_| format!("{:?}", id)), + ); + println!("{}", String::from_utf8_lossy(contents)); + } + } + } + Err(e) => { + if is_capnp_busy_error(&e) { + tracing::warn!( + operation = "poll", + total_reqs = 1, + busy_reqs = 1, + percent = 100.0, + "resource busy: ignoring" + ); + return Ok::<(), Box>(()); + } else { + return Err::<(), Box>(Box::new(e)); + } } } Ok::<(), Box>(()) @@ -207,17 +245,38 @@ async fn main() -> Result<()> { } Commands::Remove { id, lease } => { with_client(addr, |queue_client| async move { - let id_bytes = uuid::Uuid::parse_str(&id)?.into_bytes(); - let lease_bytes = uuid::Uuid::parse_str(&lease)?.into_bytes(); + let id_bytes = uuid::Uuid::parse_str(&id) + .map_err(|e| capnp::Error::failed(e.to_string()))? + .into_bytes(); + let lease_bytes = uuid::Uuid::parse_str(&lease) + .map_err(|e| capnp::Error::failed(e.to_string()))? + .into_bytes(); let mut request = queue_client.remove_request(); let mut req = request.get().init_req(); req.set_id(&id_bytes); req.set_lease(&lease_bytes); - let reply = request.send().promise.await?; - let removed = reply.get()?.get_resp()?.get_removed(); - println!("removed: {}", removed); + match request.send().promise.await { + Ok(reply) => { + let removed = reply.get()?.get_resp()?.get_removed(); + println!("removed: {}", removed); + } + Err(e) => { + if is_capnp_busy_error(&e) { + tracing::warn!( + operation = "remove", + total_reqs = 1, + busy_reqs = 1, + percent = 100.0, + "resource busy: ignoring" + ); + return Ok::<(), Box>(()); + } else { + return Err::<(), Box>(Box::new(e)); + } + } + } Ok::<(), Box>(()) }) .await? @@ -228,14 +287,33 @@ async fn main() -> Result<()> { lease_validity_secs, } => { with_client(addr, |queue_client| async move { - let lease_bytes = uuid::Uuid::parse_str(&lease)?.into_bytes(); + let lease_bytes = uuid::Uuid::parse_str(&lease) + .map_err(|e| capnp::Error::failed(e.to_string()))? + .into_bytes(); let mut request = queue_client.extend_request(); let mut req = request.get().init_req(); req.set_lease(&lease_bytes); req.set_lease_validity_secs(lease_validity_secs); - let reply = request.send().promise.await?; - let extended = reply.get()?.get_resp()?.get_extended(); - println!("extended: {}", extended); + match request.send().promise.await { + Ok(reply) => { + let extended = reply.get()?.get_resp()?.get_extended(); + println!("extended: {}", extended); + } + Err(e) => { + if is_capnp_busy_error(&e) { + tracing::warn!( + operation = "extend", + total_reqs = 1, + busy_reqs = 1, + percent = 100.0, + "resource busy: ignoring" + ); + return Ok::<(), Box>(()); + } else { + return Err::<(), Box>(Box::new(e)); + } + } + } Ok::<(), Box>(()) }) .await? @@ -256,6 +334,14 @@ async fn main() -> Result<()> { let poll_count = Arc::new(atomic::AtomicU64::new(0)); let remove_count = Arc::new(atomic::AtomicU64::new(0)); let extend_count = Arc::new(atomic::AtomicU64::new(0)); + let add_req_count = Arc::new(atomic::AtomicU64::new(0)); + let poll_req_count = Arc::new(atomic::AtomicU64::new(0)); + let remove_req_count = Arc::new(atomic::AtomicU64::new(0)); + let extend_req_count = Arc::new(atomic::AtomicU64::new(0)); + let busy_add_count = Arc::new(atomic::AtomicU64::new(0)); + let busy_poll_count = Arc::new(atomic::AtomicU64::new(0)); + let busy_remove_count = Arc::new(atomic::AtomicU64::new(0)); + let busy_extend_count = Arc::new(atomic::AtomicU64::new(0)); // periodic metrics reporter tokio::task::Builder::new() @@ -265,6 +351,14 @@ async fn main() -> Result<()> { let poll_count = Arc::clone(&poll_count); let remove_count = Arc::clone(&remove_count); let extend_count = Arc::clone(&extend_count); + let add_req_count = Arc::clone(&add_req_count); + let poll_req_count = Arc::clone(&poll_req_count); + let remove_req_count = Arc::clone(&remove_req_count); + let extend_req_count = Arc::clone(&extend_req_count); + let busy_add_count = Arc::clone(&busy_add_count); + let busy_poll_count = Arc::clone(&busy_poll_count); + let busy_remove_count = Arc::clone(&busy_remove_count); + let busy_extend_count = Arc::clone(&busy_extend_count); async move { let mut last_time = Instant::now(); while Instant::now() < end_time { @@ -274,11 +368,20 @@ async fn main() -> Result<()> { let polls = poll_count.swap(0, atomic::Ordering::Relaxed); let removes = remove_count.swap(0, atomic::Ordering::Relaxed); let extends = extend_count.swap(0, atomic::Ordering::Relaxed); + let add_reqs = add_req_count.swap(0, atomic::Ordering::Relaxed); + let poll_reqs = poll_req_count.swap(0, atomic::Ordering::Relaxed); + let remove_reqs = remove_req_count.swap(0, atomic::Ordering::Relaxed); + let extend_reqs = extend_req_count.swap(0, atomic::Ordering::Relaxed); + let busy_adds = busy_add_count.swap(0, atomic::Ordering::Relaxed); + let busy_polls = busy_poll_count.swap(0, atomic::Ordering::Relaxed); + let busy_removes = busy_remove_count.swap(0, atomic::Ordering::Relaxed); + let busy_extends = busy_extend_count.swap(0, atomic::Ordering::Relaxed); let duration = now.duration_since(last_time); last_time = now; let secs = duration.as_secs_f64().max(1.0); + let pct = |busy: u64, reqs: u64| if reqs > 0 { (busy as f64 / reqs as f64) * 100.0 } else { 0.0 }; println!( - "add: {} ({:.1}/s), poll: {} ({:.1}/s), remove: {} ({:.1}/s), extend: {} ({:.1}/s)", + "add: {} ({:.1}/s), poll: {} ({:.1}/s), remove: {} ({:.1}/s), extend: {} ({:.1}/s) | busy%% add:{:.1} poll:{:.1} remove:{:.1} extend:{:.1}", adds, adds as f64 / secs, polls, @@ -286,7 +389,11 @@ async fn main() -> Result<()> { removes, removes as f64 / secs, extends, - extends as f64 / secs + extends as f64 / secs, + pct(busy_adds, add_reqs), + pct(busy_polls, poll_reqs), + pct(busy_removes, remove_reqs), + pct(busy_extends, extend_reqs) ); } } @@ -299,6 +406,12 @@ async fn main() -> Result<()> { let poll_count = Arc::clone(&poll_count); let remove_count = Arc::clone(&remove_count); let extend_count = Arc::clone(&extend_count); + let poll_req_count = Arc::clone(&poll_req_count); + let remove_req_count = Arc::clone(&remove_req_count); + let extend_req_count = Arc::clone(&extend_req_count); + let busy_poll_count = Arc::clone(&busy_poll_count); + let busy_remove_count = Arc::clone(&busy_remove_count); + let busy_extend_count = Arc::clone(&busy_extend_count); let handle = Handle::current(); s.spawn(move || { handle.block_on(async move { @@ -313,7 +426,21 @@ async fn main() -> Result<()> { req.set_lease_validity_secs(30); req.set_num_items(10); req.set_timeout_secs(5); - let reply = request.send().promise.await.unwrap(); + poll_req_count.fetch_add(1, atomic::Ordering::Relaxed); + let reply = match request.send().promise.await { + Ok(r) => r, + Err(e) => { + if is_capnp_busy_error(&e) { + busy_poll_count.fetch_add( + 1, + atomic::Ordering::Relaxed, + ); + continue; + } else { + continue; + } + } + }; let resp = reply.get().unwrap().get_resp().unwrap(); let items = resp.get_items().unwrap(); poll_count.fetch_add( @@ -329,13 +456,23 @@ async fn main() -> Result<()> { } let promises = items.iter().map(|i| { + remove_req_count + .fetch_add(1, atomic::Ordering::Relaxed); let mut request = queue_client.remove_request(); let mut req = request.get().init_req(); req.set_id(i.get_id().unwrap()); req.set_lease(lease); request.send().promise }); - let _ = futures::future::join_all(promises).await; + let results = futures::future::join_all(promises).await; + for r in results.into_iter() { + let _ = handle_capnp_busy( + r, + "remove", + &remove_req_count, + &busy_remove_count, + ); + } remove_count.fetch_add( items.len() as u64, atomic::Ordering::Relaxed, @@ -348,7 +485,14 @@ async fn main() -> Result<()> { let mut req = request.get().init_req(); req.set_lease(&lease_arr); req.set_lease_validity_secs(30); - let _ = request.send().promise.await; + extend_req_count + .fetch_add(1, atomic::Ordering::Relaxed); + let _ = handle_capnp_busy( + request.send().promise.await, + "extend", + &extend_req_count, + &busy_extend_count, + ); extend_count .fetch_add(1, atomic::Ordering::Relaxed); } @@ -366,6 +510,8 @@ async fn main() -> Result<()> { // spawn adding clients for _ in 0..adding_clients { let add_count = Arc::clone(&add_count); + let add_req_count = Arc::clone(&add_req_count); + let busy_add_count = Arc::clone(&busy_add_count); let handle = Handle::current(); s.spawn(move || { handle.block_on(async move { @@ -390,7 +536,16 @@ async fn main() -> Result<()> { item.set_contents(format!("test {}", i).as_bytes()); item.set_visibility_timeout_secs(3); } - let _ = request.send().promise.await.unwrap(); + match handle_capnp_busy( + request.send().promise.await, + "add", + &add_req_count, + &busy_add_count, + ) { + Ok(Some(_)) => {} + Ok(None) => continue, + Err(_e) => continue, + } add_count.fetch_add( batch_size as u64, atomic::Ordering::Relaxed, @@ -435,3 +590,29 @@ where }) .await) } + +fn is_capnp_busy_error(e: &capnp::Error) -> bool { + let msg = e.to_string(); + msg.contains("Busy") || msg.contains("busy") || msg.contains("resource busy") +} + +fn handle_capnp_busy( + res: std::result::Result, + op: &str, + total_counter: &atomic::AtomicU64, + busy_counter: &atomic::AtomicU64, +) -> std::result::Result, capnp::Error> { + total_counter.fetch_add(1, atomic::Ordering::Relaxed); + match res { + Ok(v) => Ok(Some(v)), + Err(e) => { + if is_capnp_busy_error(&e) { + busy_counter.fetch_add(1, atomic::Ordering::Relaxed); + tracing::warn!(operation = op, "resource busy: ignoring"); + Ok(None) + } else { + Err(e) + } + } + } +}