diff --git a/glidefs/src/block/mod.rs b/glidefs/src/block/mod.rs index cfba221..7987045 100644 --- a/glidefs/src/block/mod.rs +++ b/glidefs/src/block/mod.rs @@ -35,10 +35,6 @@ pub mod write_trace; #[allow(unsafe_code)] // Netlink socket FFI pub mod nbd; -// Opcode policy for ublk zero-copy no-payload ops. Lives outside the -// Linux/ublk cfg so the WRITE_ZEROES-must-run contract is testable here. -pub mod ublk_zc_policy; - // ublk transport (Linux 6.0+, io_uring-based userspace block device) #[cfg(all(target_os = "linux", feature = "ublk"))] #[allow(unsafe_code)] // io_uring + eventfd FFI, single-threaded executor diff --git a/glidefs/src/block/ublk/device.rs b/glidefs/src/block/ublk/device.rs index 584a23c..bacfe50 100644 --- a/glidefs/src/block/ublk/device.rs +++ b/glidefs/src/block/ublk/device.rs @@ -13,6 +13,10 @@ use crate::block::handler::BlockHandler; use crate::task; +use ublk_core::ctrl::{UblkCtrl, UblkCtrlBuilder}; +use ublk_core::helpers::IoBuf; +use ublk_core::io::{UblkDev, UblkQueue}; +use ublk_core::{sys, BufDesc, UblkError, UblkFlags}; use std::cell::{Cell, UnsafeCell}; use std::future::Future; use std::os::unix::io::RawFd; @@ -21,10 +25,6 @@ use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::task::{Context as TaskContext, Poll, Wake, Waker}; -use ublk_core::ctrl::{UblkCtrl, UblkCtrlBuilder}; -use ublk_core::helpers::IoBuf; -use ublk_core::io::{UblkDev, UblkQueue}; -use ublk_core::{BufDesc, UblkError, UblkFlags, sys}; /// Per-queue I/O depth (max inflight commands per queue). /// @@ -228,12 +228,10 @@ impl UblkDevice { // spawn_blocking to keep the async runtime responsive. let dev_id_for_recover = dev_id; tokio::task::spawn_blocking(move || -> anyhow::Result<()> { - let ctrl = UblkCtrl::new_simple(dev_id_for_recover).map_err(|e| { - anyhow::anyhow!("UblkCtrl::new_simple({dev_id_for_recover}) failed: {e}") - })?; - ctrl.start_user_recover().map_err(|e| { - anyhow::anyhow!("start_user_recover({dev_id_for_recover}) failed: {e}") - })?; + let ctrl = UblkCtrl::new_simple(dev_id_for_recover) + .map_err(|e| anyhow::anyhow!("UblkCtrl::new_simple({dev_id_for_recover}) failed: {e}"))?; + ctrl.start_user_recover() + .map_err(|e| anyhow::anyhow!("start_user_recover({dev_id_for_recover}) failed: {e}"))?; Ok(()) }) .await??; @@ -310,7 +308,8 @@ impl UblkDevice { // // `GLIDEFS_BOUNCE_MODE=1` (test-only) disables both transports and // forces the legacy per-tag-IoBuf path. - let use_zc = features.zero_copy && std::env::var_os("GLIDEFS_BOUNCE_MODE").is_none(); + let use_zc = features.zero_copy + && std::env::var_os("GLIDEFS_BOUNCE_MODE").is_none(); if use_zc { dev_flags |= UblkFlags::UBLK_DEV_F_PREFER_ZERO_COPY; tracing::info!( @@ -394,7 +393,9 @@ impl UblkDevice { let tgt_init = move |d: &mut UblkDev| { d.tgt.dev_size = dev_size; - d.set_target_json(serde_json::json!({ "export_name": export_for_json })); + d.set_target_json( + serde_json::json!({ "export_name": export_for_json }), + ); d.tgt.params = sys::ublk_params { types: sys::UBLK_PARAM_TYPE_BASIC | sys::UBLK_PARAM_TYPE_DISCARD, basic: sys::ublk_param_basic { @@ -442,7 +443,9 @@ impl UblkDevice { qid, ready: ready_tx, }) - .map_err(|_| anyhow::anyhow!("worker {worker_idx} channel closed"))?; + .map_err(|_| { + anyhow::anyhow!("worker {worker_idx} channel closed") + })?; super::device::signal_eventfd(snap.eventfd.fd()); readys.push((*worker_idx, ready_rx)); } @@ -450,12 +453,13 @@ impl UblkDevice { for (qid_, (worker_idx, ready_rx)) in readys.into_iter().enumerate() { let t_recv = std::time::Instant::now(); - let result = ready_rx - .blocking_recv() - .map_err(|_| anyhow::anyhow!("worker {worker_idx} dropped ready sender"))?; + let result = ready_rx.blocking_recv().map_err(|_| { + anyhow::anyhow!("worker {worker_idx} dropped ready sender") + })?; timings.ready_recv_us += t_recv.elapsed().as_micros(); - result - .map_err(|s| anyhow::anyhow!("worker {worker_idx} AddQueue failed: {s}"))?; + result.map_err(|s| { + anyhow::anyhow!("worker {worker_idx} AddQueue failed: {s}") + })?; // `configure_queue` records the queue's owner thread // tid and, on the last queue, calls `build_json` — // which is what populates `/run/ublksrvd/{dev_id}.json` @@ -484,8 +488,9 @@ impl UblkDevice { // END_USER_RECOVERY for the recovery path; ublk-core's // start_dev() picks based on device state). let t = std::time::Instant::now(); - ctrl.start_dev(&dev) - .map_err(|e| anyhow::anyhow!("start_dev failed: {e:?}"))?; + ctrl.start_dev(&dev).map_err(|e| { + anyhow::anyhow!("start_dev failed: {e:?}") + })?; timings.start_dev_us = t.elapsed().as_micros(); let dev_id = i32::try_from(ctrl.dev_info().dev_id) @@ -577,9 +582,10 @@ impl UblkDevice { // send RemoveQueue messages without holding the UblkServer mutex // (the mutex serializes unrelated operations; the slow path here // is `kill_dev` which we drive outside the lock). - let worker_handles: Vec = (0..actual_nr_queues) - .map(|qid| pool.worker_snapshot(&export_name, qid, preferred_node)) - .collect(); + let worker_handles: Vec = + (0..actual_nr_queues) + .map(|qid| pool.worker_snapshot(&export_name, qid, preferred_node)) + .collect(); Ok(Self { dev_id: dev_id_assigned, @@ -680,10 +686,7 @@ impl UblkDevice { use futures::stream::{FuturesUnordered, StreamExt}; let mut remove_acks = FuturesUnordered::new(); for (qid, handle) in worker_handles.iter().enumerate() { - let key = super::worker_pool::QueueKey { - dev_id, - qid: qid as u16, - }; + let key = super::worker_pool::QueueKey { dev_id, qid: qid as u16 }; let (done_tx, done_rx) = tokio::sync::oneshot::channel(); let msg = super::worker_pool::WorkerMsg::RemoveQueue { key, done: done_tx }; match handle.inbox.try_send(msg) { @@ -696,8 +699,7 @@ impl UblkDevice { } Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { tracing::warn!( - dev_id, - qid, + dev_id, qid, "worker inbox closed during unregister (worker thread exited?)" ); } @@ -779,7 +781,10 @@ impl UblkDevice { let (done_tx, done_rx) = tokio::sync::oneshot::channel(); handle .inbox - .send(super::worker_pool::WorkerMsg::RemoveQueue { key, done: done_tx }) + .send(super::worker_pool::WorkerMsg::RemoveQueue { + key, + done: done_tx, + }) .await .map_err(|_| anyhow::anyhow!("worker channel closed for qid {qid}"))?; super::device::signal_eventfd(handle.eventfd.fd()); @@ -873,10 +878,7 @@ impl Drop for EventFd { pub(super) fn signal_eventfd(fd: RawFd) { let val: u64 = 1; let ret = unsafe { libc::write(fd, &val as *const u64 as *const libc::c_void, 8) }; - debug_assert!( - ret == 8 || ret == -1, - "eventfd write returned unexpected {ret}" - ); + debug_assert!(ret == 8 || ret == -1, "eventfd write returned unexpected {ret}"); } /// Drain accumulated eventfd signals (non-blocking read). @@ -884,10 +886,7 @@ pub(super) fn drain_eventfd(fd: RawFd) { let mut val: u64 = 0; let ret = unsafe { libc::read(fd, &mut val as *mut u64 as *mut libc::c_void, 8) }; // EAGAIN is expected when no signals are pending (EFD_NONBLOCK). - debug_assert!( - ret == 8 || ret == -1, - "eventfd read returned unexpected {ret}" - ); + debug_assert!(ret == 8 || ret == -1, "eventfd read returned unexpected {ret}"); } // --------------------------------------------------------------------------- @@ -918,10 +917,7 @@ impl WakeupBits { pub(super) fn new(num_words: usize, efd: Arc) -> Self { assert!(num_words > 0, "WakeupBits needs at least one word"); let words: Vec = (0..num_words).map(|_| AtomicU64::new(0)).collect(); - Self { - words: words.into_boxed_slice(), - efd, - } + Self { words: words.into_boxed_slice(), efd } } /// Total task index capacity (number of bits). @@ -1094,15 +1090,13 @@ impl<'a> QueueExecutor<'a> { /// the event loop exit. pub(super) fn spawn_daemon(&mut self, future: impl Future + 'a) { debug_assert_eq!( - self.alive.get(), - 0, + self.alive.get(), 0, "spawn_daemon must be called before spawn" ); let idx = self.tasks.len(); assert!( idx < self.bits.capacity(), - "QueueExecutor capacity {} exceeded by spawn_daemon", - self.bits.capacity() + "QueueExecutor capacity {} exceeded by spawn_daemon", self.bits.capacity() ); self.tasks.push(UnsafeCell::new(Some(Box::pin(future)))); self.wakers.push(Waker::from(Arc::new(TaskWaker { @@ -1244,6 +1238,7 @@ fn panic_payload_to_str(payload: &Box) -> &str { } } + /// Per-tag async I/O task. Runs in a worker's [`QueueExecutor`] and /// pumps the FETCH → dispatch → COMMIT cycle for one tag's lifetime. /// @@ -1427,14 +1422,6 @@ impl ublk_core::zc::ZcTarget for GlidefsZcTarget { use ublk_core::zc::{ZcAction, ZcChunk, ZcChunkOp, ZcDispatch}; match u32::from(op) { - op if crate::block::ublk_zc_policy::zc_no_payload(op) - == Some(crate::block::ublk_zc_policy::ZcNoPayload::AdvertisedNoop) => - { - ZcDispatch::Inline { - action: ZcAction::Complete(0), - keepalive: None, - } - } sys::UBLK_IO_OP_FLUSH => { // FLUSH must be Deferred, not Inline. `handler.flush()` → // `cache.flush()` acquires `data_file.read()` task-fairly. @@ -1465,21 +1452,8 @@ impl ublk_core::zc::ZcTarget for GlidefsZcTarget { }); ZcDispatch::Deferred } - sys::UBLK_IO_OP_WRITE_ZEROES => { - // Advertised (max_write_zeroes_sectors = 16 MiB). The USER_COPY - // path calls handler.write_zeroes; ZC used to Complete(0) and - // leave guest-visible stale bytes. Run the same handler. - let handler = Arc::clone(&self.handler); - let handle = handle.clone(); - self.runtime.spawn(async move { - let mut guard = SubmitGuard::new(tag, handle); - let res = run_mutating("ublk-zc-write-zeroes", &handler, || { - handler.write_zeroes(offset, length, false) - }) - .await; - guard.commit(ZcAction::Complete(res), None); - }); - ZcDispatch::Deferred + sys::UBLK_IO_OP_DISCARD | sys::UBLK_IO_OP_WRITE_ZEROES => { + ZcDispatch::Inline { action: ZcAction::Complete(0), keepalive: None } } sys::UBLK_IO_OP_WRITE => { // Inline fast path: avoid `runtime.spawn` + mpsc/eventfd @@ -1520,10 +1494,7 @@ impl ublk_core::zc::ZcTarget for GlidefsZcTarget { ) { Ok(()) => { let action = ZcAction::Chunks(vec![ZcChunk { - op: ZcChunkOp::WriteFixed { - fd, - dst_offset: offset, - }, + op: ZcChunkOp::WriteFixed { fd, dst_offset: offset }, buf_offset: 0, length, }]); @@ -1582,10 +1553,7 @@ impl ublk_core::zc::ZcTarget for GlidefsZcTarget { use std::os::unix::io::AsRawFd; let fd = (*gate).as_raw_fd_for_ublk_zc(); let action = ZcAction::Chunks(vec![ZcChunk { - op: ZcChunkOp::WriteFixed { - fd, - dst_offset: offset, - }, + op: ZcChunkOp::WriteFixed { fd, dst_offset: offset }, buf_offset: 0, length, }]); @@ -1642,7 +1610,8 @@ impl ublk_core::zc::ZcTarget for GlidefsZcTarget { // plane window (pwrite + READ_FIXED). { let gate = handler.zc_inflight_enter(); - if let Some(action) = try_zc_read_hot_path(&handler, &*gate, offset, length) + if let Some(action) = + try_zc_read_hot_path(&handler, &*gate, offset, length) { guard.commit(action, Some(Box::new(gate))); return; @@ -1674,10 +1643,7 @@ impl ublk_core::zc::ZcTarget for GlidefsZcTarget { }); ZcDispatch::Deferred } - _ => ZcDispatch::Inline { - action: ZcAction::Complete(-libc::EIO), - keepalive: None, - }, + _ => ZcDispatch::Inline { action: ZcAction::Complete(-libc::EIO), keepalive: None }, } } @@ -1706,17 +1672,10 @@ impl ublk_core::zc::ZcTarget for GlidefsZcTarget { }) .map(|g| &**g); let Some(df) = df else { - tracing::error!( - offset, - length, - "ZC after_write: missing rotation-gate keepalive" - ); + tracing::error!(offset, length, "ZC after_write: missing rotation-gate keepalive"); return -libc::EIO; }; - match self - .handler - .commit_after_zc_write_with(df, offset, u64::from(length), false) - { + match self.handler.commit_after_zc_write_with(df, offset, u64::from(length), false) { Ok(()) => { #[allow(clippy::cast_possible_wrap)] let r = length as i32; @@ -1742,10 +1701,7 @@ struct SubmitGuard { impl SubmitGuard { fn new(tag: u16, handle: ublk_core::zc::ZcQueueHandle) -> Self { - Self { - tag, - handle: Some(handle), - } + Self { tag, handle: Some(handle) } } fn commit( @@ -1766,11 +1722,7 @@ impl Drop for SubmitGuard { tag = self.tag, "ZC dispatch task dropped without commit — sending -EIO" ); - handle.submit( - self.tag, - ublk_core::zc::ZcAction::Complete(-libc::EIO), - None, - ); + handle.submit(self.tag, ublk_core::zc::ZcAction::Complete(-libc::EIO), None); } } } @@ -1821,9 +1773,7 @@ fn try_zc_read_hot_path( let chunk_start_byte = block_idx * block_size; let slice_start = if block_idx == start_block { #[allow(clippy::cast_possible_truncation)] - { - (offset - chunk_start_byte) as u32 - } + { (offset - chunk_start_byte) as u32 } } else { 0 }; @@ -1992,8 +1942,7 @@ pub(super) async fn io_task_zero_copy( // on zero-block chunks — kernel reads zeros directly into bio, no // userspace memset, no per-tag scratch buffer. let dev_zero_path = std::ffi::CString::new("/dev/zero").unwrap(); - let dev_zero_fd = - unsafe { libc::open(dev_zero_path.as_ptr(), libc::O_RDONLY | libc::O_CLOEXEC) }; + let dev_zero_fd = unsafe { libc::open(dev_zero_path.as_ptr(), libc::O_RDONLY | libc::O_CLOEXEC) }; if dev_zero_fd < 0 { return Err(UblkError::IOError(std::io::Error::last_os_error())); } @@ -2155,8 +2104,7 @@ async fn io_task_user_copy( let qid = q.get_qid(); // Initial fetch — no buffer attached, empty slice. - q.submit_io_prep_cmd(tag, BufDesc::Slice(&[]), 0, None) - .await?; + q.submit_io_prep_cmd(tag, BufDesc::Slice(&[]), 0, None).await?; loop { let iod = q.get_iod(tag); @@ -2194,26 +2142,20 @@ async fn io_task_user_copy( let Some(mut iobuf) = super::buffer_pool::acquire_io_buf(length as usize).await else { tracing::error!( - qid, - tag, - length, + qid, tag, length, max_len = super::buffer_pool::SLOT_SIZE, "bounce buffer unavailable or length invalid — failing this I/O with EIO", ); - q.submit_io_commit_cmd(tag, BufDesc::Slice(&[]), -libc::EIO) - .await?; + q.submit_io_commit_cmd(tag, BufDesc::Slice(&[]), -libc::EIO).await?; continue; }; let Some(buf) = iobuf.as_mut_slice(length as usize) else { tracing::error!( - qid, - tag, - length, + qid, tag, length, max_len = super::buffer_pool::SLOT_SIZE, "bounce buffer length exceeds slot size — failing this I/O with EIO", ); - q.submit_io_commit_cmd(tag, BufDesc::Slice(&[]), -libc::EIO) - .await?; + q.submit_io_commit_cmd(tag, BufDesc::Slice(&[]), -libc::EIO).await?; continue; }; @@ -2236,13 +2178,7 @@ async fn io_task_user_copy( }; if ret < 0 { let err = std::io::Error::last_os_error(); - tracing::error!( - ?err, - qid, - tag, - length, - "USER_COPY pread WRITE-data failed" - ); + tracing::error!(?err, qid, tag, length, "USER_COPY pread WRITE-data failed"); -err.raw_os_error().unwrap_or(libc::EIO) } else { handle_io(op, offset, length, fua, buf, handler).await @@ -2261,13 +2197,7 @@ async fn io_task_user_copy( }; if ret < 0 { let err = std::io::Error::last_os_error(); - tracing::error!( - ?err, - qid, - tag, - res, - "USER_COPY pwrite READ-data failed" - ); + tracing::error!(?err, qid, tag, res, "USER_COPY pwrite READ-data failed"); -err.raw_os_error().unwrap_or(libc::EIO) } else { res @@ -2285,8 +2215,7 @@ async fn io_task_user_copy( _ => -libc::EINVAL, }; - q.submit_io_commit_cmd(tag, BufDesc::Slice(&[]), result) - .await?; + q.submit_io_commit_cmd(tag, BufDesc::Slice(&[]), result).await?; } } @@ -2401,8 +2330,7 @@ async fn handle_io( // Run through the panic-isolating wrapper so a write_cache // panic marks the export degraded instead of poisoning peer // queues (parallel to NBD's run_mutating_request). - let result = - run_mutating("ublk-write", handler, || handler.write(offset, buf, fua)).await; + let result = run_mutating("ublk-write", handler, || handler.write(offset, buf, fua)).await; if result == 0 { i32::try_from(length).unwrap_or(-libc::EIO) } else { @@ -2471,10 +2399,9 @@ mod tests { let clean_cache: Arc = Arc::new(SimpleBlockCache::new(64 * 1024 * 1024)); let pack_index_cache = Arc::new(PackIndexCache::open(temp.path()).await.unwrap()); - let volume_manifest = Arc::new(parking_lot::RwLock::new(VolumeManifest::new( - DEVICE_SIZE, - BLOCK_SIZE as u32, - ))); + let volume_manifest = Arc::new(parking_lot::RwLock::new( + VolumeManifest::new(DEVICE_SIZE, BLOCK_SIZE as u32), + )); let metrics = Arc::new(ExportMetrics::new()); let cache = WriteCache::open(config).unwrap().skip_recovery_for_test(); let handler = BlockHandler::new( @@ -2502,27 +2429,13 @@ mod tests { let (handler, _dir) = make_handler(false).await; let mut buf = vec![0x42u8; BLOCK_SIZE]; - let result = handle_io( - ublk_core::sys::UBLK_IO_OP_WRITE, - 0, - BLOCK_SIZE as u32, - false, - &mut buf, - &handler, - ) - .await; + let result = + handle_io(ublk_core::sys::UBLK_IO_OP_WRITE, 0, BLOCK_SIZE as u32, false, &mut buf, &handler).await; assert_eq!(result, BLOCK_SIZE as i32); let mut buf = vec![0u8; BLOCK_SIZE]; - let result = handle_io( - ublk_core::sys::UBLK_IO_OP_READ, - 0, - BLOCK_SIZE as u32, - false, - &mut buf, - &handler, - ) - .await; + let result = + handle_io(ublk_core::sys::UBLK_IO_OP_READ, 0, BLOCK_SIZE as u32, false, &mut buf, &handler).await; assert_eq!(result, BLOCK_SIZE as i32); assert_eq!(buf, vec![0x42u8; BLOCK_SIZE]); } @@ -2530,87 +2443,35 @@ mod tests { #[tokio::test] async fn flush_returns_ok() { let (handler, _dir) = make_handler(false).await; - let result = handle_io( - ublk_core::sys::UBLK_IO_OP_FLUSH, - 0, - 0, - false, - &mut [], - &handler, - ) - .await; + let result = + handle_io(ublk_core::sys::UBLK_IO_OP_FLUSH, 0, 0, false, &mut [], &handler).await; assert_eq!(result, 0); } #[tokio::test] async fn discard_returns_ok() { let (handler, _dir) = make_handler(false).await; - let result = handle_io( - ublk_core::sys::UBLK_IO_OP_DISCARD, - 0, - BLOCK_SIZE as u32, - false, - &mut [], - &handler, - ) - .await; + let result = + handle_io(ublk_core::sys::UBLK_IO_OP_DISCARD, 0, BLOCK_SIZE as u32, false, &mut [], &handler).await; assert_eq!(result, 0); } - #[test] - fn zc_policy_opcodes_match_sys() { - use crate::block::ublk_zc_policy::{self, ZcNoPayload}; - assert_eq!(sys::UBLK_IO_OP_FLUSH, ublk_zc_policy::UBLK_IO_OP_FLUSH); - assert_eq!(sys::UBLK_IO_OP_DISCARD, ublk_zc_policy::UBLK_IO_OP_DISCARD); - assert_eq!( - sys::UBLK_IO_OP_WRITE_ZEROES, - ublk_zc_policy::UBLK_IO_OP_WRITE_ZEROES - ); - assert_eq!( - ublk_zc_policy::zc_no_payload(sys::UBLK_IO_OP_WRITE_ZEROES), - Some(ZcNoPayload::RunHandler) - ); - } - #[tokio::test] async fn write_zeroes_clears_data() { let (handler, _dir) = make_handler(false).await; // Write non-zero data. let mut buf = vec![0xFFu8; BLOCK_SIZE]; - handle_io( - ublk_core::sys::UBLK_IO_OP_WRITE, - 0, - BLOCK_SIZE as u32, - false, - &mut buf, - &handler, - ) - .await; + handle_io(ublk_core::sys::UBLK_IO_OP_WRITE, 0, BLOCK_SIZE as u32, false, &mut buf, &handler).await; // Write zeroes over it. - let result = handle_io( - ublk_core::sys::UBLK_IO_OP_WRITE_ZEROES, - 0, - BLOCK_SIZE as u32, - false, - &mut [], - &handler, - ) - .await; + let result = + handle_io(ublk_core::sys::UBLK_IO_OP_WRITE_ZEROES, 0, BLOCK_SIZE as u32, false, &mut [], &handler).await; assert_eq!(result, 0); // Read back — should be zeros. let mut buf = vec![0xFFu8; BLOCK_SIZE]; - handle_io( - ublk_core::sys::UBLK_IO_OP_READ, - 0, - BLOCK_SIZE as u32, - false, - &mut buf, - &handler, - ) - .await; + handle_io(ublk_core::sys::UBLK_IO_OP_READ, 0, BLOCK_SIZE as u32, false, &mut buf, &handler).await; assert_eq!(buf, vec![0u8; BLOCK_SIZE]); } @@ -2644,15 +2505,8 @@ mod tests { // BlockHandler — confirms the protection actually fires for ublk // requests, not just NBD. let mut buf = vec![0x42u8; BLOCK_SIZE]; - let next = handle_io( - ublk_core::sys::UBLK_IO_OP_WRITE, - 0, - BLOCK_SIZE as u32, - false, - &mut buf, - &handler, - ) - .await; + let next = + handle_io(ublk_core::sys::UBLK_IO_OP_WRITE, 0, BLOCK_SIZE as u32, false, &mut buf, &handler).await; assert_eq!(next, -libc::EIO); } @@ -2673,19 +2527,9 @@ mod tests { async fn read_beyond_device_returns_zeros() { let (handler, _dir) = make_handler(false).await; let mut buf = vec![0xFFu8; BLOCK_SIZE]; - let result = handle_io( - ublk_core::sys::UBLK_IO_OP_READ, - DEVICE_SIZE, - BLOCK_SIZE as u32, - false, - &mut buf, - &handler, - ) - .await; - assert_eq!( - result, BLOCK_SIZE as i32, - "OOB read should succeed with zero-fill" - ); + let result = + handle_io(ublk_core::sys::UBLK_IO_OP_READ, DEVICE_SIZE, BLOCK_SIZE as u32, false, &mut buf, &handler).await; + assert_eq!(result, BLOCK_SIZE as i32, "OOB read should succeed with zero-fill"); assert!(buf.iter().all(|&b| b == 0), "OOB read should return zeros"); } @@ -2693,15 +2537,8 @@ mod tests { async fn write_with_fua_succeeds() { let (handler, _dir) = make_handler(false).await; let mut buf = vec![0xABu8; BLOCK_SIZE]; - let result = handle_io( - ublk_core::sys::UBLK_IO_OP_WRITE, - 0, - BLOCK_SIZE as u32, - true, - &mut buf, - &handler, - ) - .await; + let result = + handle_io(ublk_core::sys::UBLK_IO_OP_WRITE, 0, BLOCK_SIZE as u32, true, &mut buf, &handler).await; assert_eq!(result, BLOCK_SIZE as i32); } @@ -2709,30 +2546,16 @@ mod tests { async fn write_readonly_returns_erofs() { let (handler, _dir) = make_handler(true).await; let mut buf = vec![0x42u8; BLOCK_SIZE]; - let result = handle_io( - ublk_core::sys::UBLK_IO_OP_WRITE, - 0, - BLOCK_SIZE as u32, - false, - &mut buf, - &handler, - ) - .await; + let result = + handle_io(ublk_core::sys::UBLK_IO_OP_WRITE, 0, BLOCK_SIZE as u32, false, &mut buf, &handler).await; assert_eq!(result, -libc::EROFS); } #[tokio::test] async fn zero_length_read_returns_zero() { let (handler, _dir) = make_handler(false).await; - let result = handle_io( - ublk_core::sys::UBLK_IO_OP_READ, - 0, - 0, - false, - &mut [], - &handler, - ) - .await; + let result = + handle_io(ublk_core::sys::UBLK_IO_OP_READ, 0, 0, false, &mut [], &handler).await; assert_eq!(result, 0); } diff --git a/glidefs/src/block/ublk_zc_policy.rs b/glidefs/src/block/ublk_zc_policy.rs deleted file mode 100644 index 9d01289..0000000 --- a/glidefs/src/block/ublk_zc_policy.rs +++ /dev/null @@ -1,65 +0,0 @@ -//! Policy for ublk no-payload ops on the zero-copy dispatch path. -//! -//! Extracted so Darwin (and any host without ublk) can still lock the -//! contract: `WRITE_ZEROES` is advertised (`max_write_zeroes_sectors` is -//! non-zero) and must run the handler. `DISCARD` is advertised as -//! unsupported (`max_discard_sectors = 0`); ACK-0 is correct for that -//! opcode only. -//! -//! Opcode numbers are the Linux `ublk_cmd.h` values. A Linux-only test -//! next to the ublk device asserts they still match `ublk_core::sys`. - -/// `UBLK_IO_OP_FLUSH` (`ublk_cmd.h`). -pub const UBLK_IO_OP_FLUSH: u32 = 2; -/// `UBLK_IO_OP_DISCARD` (`ublk_cmd.h`). -pub const UBLK_IO_OP_DISCARD: u32 = 3; -/// `UBLK_IO_OP_WRITE_ZEROES` (`ublk_cmd.h`). -pub const UBLK_IO_OP_WRITE_ZEROES: u32 = 5; - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ZcNoPayload { - /// Run the handler; do not ACK until it completes. - RunHandler, - /// The device advertised this opcode as unsupported. ACK 0 is correct. - AdvertisedNoop, -} - -/// What the ZC dispatch must do for a no-payload ublk opcode. -pub fn zc_no_payload(op: u32) -> Option { - match op { - UBLK_IO_OP_FLUSH => Some(ZcNoPayload::RunHandler), - UBLK_IO_OP_DISCARD => Some(ZcNoPayload::AdvertisedNoop), - UBLK_IO_OP_WRITE_ZEROES => Some(ZcNoPayload::RunHandler), - _ => None, - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn write_zeroes_must_run_the_handler() { - assert_eq!( - zc_no_payload(UBLK_IO_OP_WRITE_ZEROES), - Some(ZcNoPayload::RunHandler), - "ZC WRITE_ZEROES is advertised; ACK-0 without zeroing is silent wrong" - ); - } - - #[test] - fn discard_is_an_advertised_noop() { - assert_eq!( - zc_no_payload(UBLK_IO_OP_DISCARD), - Some(ZcNoPayload::AdvertisedNoop) - ); - } - - #[test] - fn flush_must_run_the_handler() { - assert_eq!( - zc_no_payload(UBLK_IO_OP_FLUSH), - Some(ZcNoPayload::RunHandler) - ); - } -}