From af0b3737d1d89d608fe63d2e6179c4c2c387c01c Mon Sep 17 00:00:00 2001 From: Zhiheng Tao Date: Thu, 30 Jul 2026 15:24:13 +0800 Subject: [PATCH] feat(uffd): update backend protocol Update handshake flags, UFFD modes, and region metadata. Handle managed faults and acknowledgements. Accept and close unused backing FDs. Signed-off-by: Zhiheng Tao --- docs/uffd.md | 229 ++++++++++++-------- nydus/src/uffd/core.rs | 27 ++- nydus/src/uffd/mod.rs | 2 +- nydus/src/uffd/proto.rs | 445 +++++++++++++++++++++++--------------- nydus/src/uffd/service.rs | 43 ++-- 5 files changed, 442 insertions(+), 304 deletions(-) diff --git a/docs/uffd.md b/docs/uffd.md index b535ffe5c3d..c21720a409f 100644 --- a/docs/uffd.md +++ b/docs/uffd.md @@ -33,23 +33,24 @@ anonymous virtio-pmem mapping v local cache files -Zerocopy: nydus uffd -- FD ranges --> microVM process - | - mmap(MAP_FIXED) + UFFD wake - | - v - guest thread resumes - -Copy: nydus uffd -- UFFDIO_COPY / UFFDIO_ZEROPAGE --> guest thread resumes +Customized (zero-copy): nydus uffd -- FD ranges --> microVM process + | + mmap(MAP_FIXED) + UFFD wake + | + v + guest thread resumes + +Managed (copy): nydus uffd -- UFFDIO_COPY / UFFDIO_ZEROPAGE --> guest thread resumes ``` Two fault policies are supported: -- **Zerocopy**: Nydus returns file descriptors and byte ranges. The microVM - maps those ranges over the anonymous virtio-pmem VMA and wakes the faulting - thread. -- **Copy**: Nydus reads the resolved bytes and completes the fault itself with - `UFFDIO_COPY` or `UFFDIO_ZEROPAGE`. No range response is sent. +- **Customized**: Nydus returns file descriptors and byte ranges for the + zero-copy path. The microVM maps those ranges over the anonymous + virtio-pmem VMA and wakes the faulting thread. +- **Managed**: Nydus reads the resolved bytes and completes the fault itself + through the copy path with `UFFDIO_COPY` or `UFFDIO_ZEROPAGE`. No range + response is sent. The service also supports stateless `STAT`, `FETCH`, and `PROBE` requests for a client that monitors its own userfaultfd. @@ -88,7 +89,7 @@ fetched, decoded, and validated before its file descriptor is returned. ## Transport and Framing The server listens on an `AF_UNIX`, `SOCK_STREAM` socket. Every frame consists -of a fixed 20-byte header followed by `len` payload bytes. All integer fields +of a fixed 16-byte header followed by `len` payload bytes. All integer fields are little-endian. Because the transport is a stream, clients must not assume that one `sendmsg` @@ -99,65 +100,93 @@ File descriptors are attached to the frame header with `SCM_RIGHTS`. They are not included in `header.len`. A receiver owns every received descriptor and must close it after use. -### Common header +### Request header | Offset | Size | Type | Field | Meaning | |---:|---:|---|---|---| -| 0 | 4 | `u32` | `magic` | Always `0x55464644` (`UFFD`) | -| 4 | 2 | `u16` | `flags` | Message-specific flags | -| 6 | 2 | `u16` | `msg_type` | Request or response type | -| 8 | 8 | `u64` | `cookie` | Reserved for correlation/extension | -| 16 | 4 | `u32` | `len` | Payload length, excluding header and FDs | - -`RANGE_RESPONSE` defines a `NEXT` bit in `flags`; other messages currently set -`flags` to zero. The receiver accepts unknown flag bits for future extension. -`cookie` is not currently interpreted and should be zero. - -The implementation bounds accepted payload lengths. An invalid magic value, -malformed payload, invalid FD count, or payload exceeding that bound terminates -the connection. Unknown message types are logged and ignored. - -## Message Types - -| Value | Name | Direction | Connection state | -|---:|---|---|---| -| `0x01` | `HANDSHAKE` | client to server | Establishes UFFD state | -| `0x02` | `STAT_REQUEST` | client to server | Stateless | -| `0x03` | `FETCH_REQUEST` | client to server | Stateless | -| `0x04` | `PROBE_REQUEST` | client to server | Stateless | -| `0x81` | `RANGE_RESPONSE` | server to client | Backing file ranges | -| `0x82` | `STAT_RESPONSE` | server to client | Stat response | - -There are no dynamic add-region or remove-region messages. A connection sends -its complete region list in `HANDSHAKE`, which may occur at most once. +| 0 | 2 | `u16` | `command` | Request command | +| 2 | 6 | `u8[6]` | `command_headers` | Command-specific fields | +| 8 | 8 | `u64` | `len` | Payload length, excluding header and FDs | + +Commands that do not currently define command-specific fields send these +bytes as zero. The server ignores unused bytes so future extensions remain +backward compatible. + +### Response header + +| Offset | Size | Type | Field | Meaning | +|---:|---:|---|---|---| +| 0 | 2 | `u16` | `status` | `OK=1`, `ERROR=2` | +| 2 | 2 | `u16` | `reply_type` | Response payload type | +| 4 | 4 | `u8[4]` | `reply_headers` | Reply-specific fields | +| 8 | 8 | `u64` | `len` | Payload length, excluding header and FDs | + +The implementation bounds accepted payload lengths. A malformed payload, +invalid FD count, or payload exceeding that bound terminates the connection. +Unknown commands are logged and ignored. + +## Commands and Reply Types + +| Value | Command | Connection state | +|---:|---|---| +| `0x0a` | `HANDSHAKE` | Establishes UFFD state | +| `0x0b` | `ADD_REGION` | Not implemented by this service | +| `0x0c` | `REMOVE_REGION` | Not implemented by this service | +| `0x20` | `STAT` | Stateless | +| `0x21` | `FETCH` | Stateless | +| `0x22` | `PROBE` | Stateless | + +| Value | Reply type | Meaning | +|---:|---|---| +| `0` | `LEGACY` | Legacy inline or empty response | +| `1` | `FD_RANGES` | Backing file ranges | +| `0x20` | `STAT` | Device metadata | + +Nydus does not currently implement dynamic add-region or remove-region +requests. A connection sends its complete region list in `HANDSHAKE`, which may +occur at most once. ## HANDSHAKE `HANDSHAKE` registers one userfaultfd and the VMAs backed by the flattened device. -Exactly one userfaultfd must be attached with `SCM_RIGHTS`. +The first FD attached with `SCM_RIGHTS` is the userfaultfd. Optional backing +FDs are accepted and closed without use, as described below. -### Payload prefix +### Command header | Offset | Size | Type | Field | Meaning | |---:|---:|---|---|---| | 0 | 2 | `u16` | `version` | Must be `1` | -| 2 | 1 | `u8` | `flags` | Fault policy and prefault flags | -| 3 | 1 | `u8` | `region_count` | Number of following regions | +| 2 | 1 | `u8` | `flags` | Handshake behavior flags | +| 3 | 1 | `u8` | `uffd_modes` | Effective client UFFD registration modes | +| 4 | 2 | `u16` | `region_count` | Number of payload regions | Handshake flags: | Bit | Name | Meaning | |---:|---|---| -| 0 | `COPY` | Use Copy policy; clear means Zerocopy | +| 0 | `MANAGED` | Nydus handles faults using the copy path; clear selects the Customized zero-copy path | | 1 | `PREFAULT` | Send locally ready ranges during handshake | +| 2 | `ACK_REQUIRED` | Return an empty Legacy response after acceptance | +| 3 | `BACKING_FDS` | One backing FD follows the userfaultfd per region | Other flag bits are currently ignored. +UFFD modes: + +| Bit | Name | Meaning | +|---:|---|---| +| 0 | `MISSING` | Regions use UFFD missing mode | +| 1 | `WP` | Regions use UFFD write-protect mode | +| 2 | `WP_ASYNC` | UFFD write protection is asynchronous | + +Nydus currently ignores `uffd_modes`, including unknown mode bits. + ### Region entry -Each region is 40 bytes. +Each region is 48 bytes. | Offset | Size | Type | Field | Meaning | |---:|---:|---|---|---| @@ -167,33 +196,49 @@ Each region is 40 bytes. | 24 | 8 | `u64` | `fault_size` | Fault resolution window | | 32 | 4 | `i32` | `prot` | Mapping protection flags | | 36 | 4 | `i32` | `flags` | Mapping flags | +| 40 | 8 | `u64` | `backing_offset` | Region offset in its backing FD | -Region entries immediately follow the 4-byte prefix. Payload length is: +Region entries start at the beginning of the payload. Payload length is: ```text -4 + region_count * 40 +region_count * 48 ``` +Without `BACKING_FDS`, the request carries only the userfaultfd. With +`BACKING_FDS`, it carries the userfaultfd followed by one backing FD per region. +Nydus validates this ordering and count, retains the userfaultfd, and +immediately closes the backing FDs because its current fault paths do not use +them. `backing_offset` is parsed but otherwise ignored. + The service makes the received userfaultfd nonblocking and monitors it for page-fault events. A duplicate handshake is rejected. -When Zerocopy prefault is enabled, Nydus probes the registered regions and -sends their currently ready ranges as `RANGE_RESPONSE` frames. Prefault does -not download missing blob data. It is processed synchronously so range-response -frames are not written concurrently on the stream. Copy policy ignores the -prefault flag. +When `ACK_REQUIRED` is set, Nydus sends an empty Legacy `OK` response after +accepting the session and before sending any prefault `FD_RANGES`. Without the +flag, a successful handshake has no response. Handshake failure may close the +connection without an error response. -## RANGE_RESPONSE +When Customized prefault is enabled, Nydus probes the registered regions and +sends their currently ready ranges as `FD_RANGES` responses. Prefault does not +download missing blob data. It is processed synchronously so range frames are +not written concurrently on the stream. Managed policy ignores the prefault +flag. -`RANGE_RESPONSE` returns backing file ranges for Zerocopy faults, prefault, +## FD_RANGES + +`FD_RANGES` returns backing file ranges for Customized faults, prefault, `FETCH`, and `PROBE`. -### Payload +### Reply header | Offset | Size | Type | Field | Meaning | |---:|---:|---|---|---| -| 0 | 4 | `u32` | `range_count` | Number of range entries and attached FDs | -| 4 | variable | `Range[]` | `ranges` | `range_count` 24-byte entries | +| 0 | 2 | `u16` | `flags` | Bit zero is `MORE` | +| 2 | 2 | `u16` | `fd_count` | Number of payload entries and attached FDs | + +### Payload + +The payload contains exactly `fd_count` consecutive 24-byte range entries. Each range entry is: @@ -205,27 +250,27 @@ Each range entry is: One FD is attached for every range, in the same order as the entries. The implementation limits the number of ranges and FDs in one frame. Larger -results are split across multiple `RANGE_RESPONSE` frames, so clients must not +results are split across multiple `FD_RANGES` frames, so clients must not assume a fixed batch size. -Bit zero of the RANGE_RESPONSE header `flags` field is `NEXT`: +Bit zero of the reply-specific `flags` field is `MORE`: -- `NEXT=1` means another RANGE_RESPONSE for the same logical result follows. -- `NEXT=0` means the current frame is the final batch. -- A single-batch result has `NEXT=0`. -- An empty result is one RANGE_RESPONSE with `range_count=0`, no FDs, and - `NEXT=0`. +- `MORE=1` means another `FD_RANGES` batch for the same logical result follows. +- `MORE=0` means the current frame is the final batch. +- A single-batch result has `MORE=0`. +- An empty result is one `FD_RANGES` response with `fd_count=0`, no FDs, and + `MORE=0`. -These rules apply to Zerocopy faults, handshake prefault, `FETCH`, and `PROBE`. -All batches for one result are sent contiguously and are not interleaved with -another result on the same connection. +These rules apply to Customized faults, handshake prefault, `FETCH`, and +`PROBE`. All batches for one result are sent contiguously and are not +interleaved with another result on the same connection. Data ranges reference either the bootstrap or a decoded blob cache file. Hole ranges reference `/dev/zero` with `file_offset == 0`. Adjacent ranges are merged when they use the same FD and contiguous device/file offsets; adjacent hole ranges may also be merged. -For a Zerocopy fault, the client maps each range at: +For a Customized fault, the client maps each range at: ```text host_address = region.virt_addr @@ -238,10 +283,11 @@ range and closes the received FDs. ## STAT -`STAT_REQUEST` has an empty payload and carries no FDs. It may be sent on a -connection without a handshake. +`STAT` currently sends zeroed command-specific headers, an empty payload, and +no FDs. It may be sent on a connection without a handshake. -The server replies with one `STAT_RESPONSE` and no FDs: +The server replies with one `STAT` response and no FDs. Its reply-specific +headers are zero and its payload is: | Offset | Size | Type | Field | Meaning | |---:|---:|---|---|---| @@ -251,8 +297,9 @@ The server replies with one `STAT_RESPONSE` and no FDs: ## FETCH -`FETCH_REQUEST` asks Nydus to make one device range locally available and -return its complete FD mapping. It has no connection state and carries no FDs. +`FETCH` asks Nydus to make one device range locally available and return its +complete FD mapping. Its command-specific headers are currently zero, it has +no connection state, and it carries no FDs. | Offset | Size | Type | Field | Meaning | |---:|---:|---|---|---| @@ -260,25 +307,26 @@ return its complete FD mapping. It has no connection state and carries no FDs. | 8 | 8 | `u64` | `len` | Nonzero, block-aligned byte length | Nydus downloads and decodes missing blob groups, then returns one or more -`RANGE_RESPONSE` frames. The returned ranges cover the complete requested +`FD_RANGES` frames. The returned ranges cover the complete requested interval without gaps; holes are represented by `/dev/zero` ranges. The client -receives through the RANGE_RESPONSE with `NEXT=0`, then verifies that the +receives through the `FD_RANGES` response with `MORE=0`, then verifies that the accumulated ranges cover `[offset, offset + len)`. ## PROBE -`PROBE_REQUEST` has an empty payload, carries no FDs, and requires no -handshake. It checks the entire flattened device without downloading missing -blob data. +`PROBE` currently sends zeroed command-specific headers and an empty payload, +carries no FDs, and requires no handshake. It checks the entire flattened +device without downloading missing blob data. -The server emits one or more `RANGE_RESPONSE` frames containing: +The server emits one or more `FD_RANGES` frames containing: - the bootstrap range; - hole ranges backed by `/dev/zero`; - blob subranges already present in the local cache. -Missing blob ranges are omitted. The RANGE_RESPONSE with `NEXT=0` completes the -probe. If no ranges are ready, that final response has zero ranges and no FDs. +Missing blob ranges are omitted. The `FD_RANGES` response with `MORE=0` +completes the probe. If no ranges are ready, that final response has zero +ranges and no FDs. ## Fault Handling @@ -287,19 +335,19 @@ fault address. It aligns the fault offset down to the region's `fault_size` (with a minimum of the service block size) and clips the resolution window to the region end. -### Zerocopy policy +### Customized policy (zero-copy) 1. Fetch and validate every blob range in the fault window. 2. Resolve the window into bootstrap/blob/zero FD ranges. -3. Send one or more `RANGE_RESPONSE` frames. +3. Send one or more `FD_RANGES` frames. 4. The client installs fixed mappings and wakes the faulting thread. -### Copy policy +### Managed policy (copy) 1. Fetch and validate every blob range in the fault window. 2. Read data ranges from their backing FDs. 3. Resolve data with `UFFDIO_COPY` and holes with `UFFDIO_ZEROPAGE`. -4. Return to the connection loop without sending `RANGE_RESPONSE`. +4. Return to the connection loop without sending `FD_RANGES`. Faults are processed serially within one connection. Separate connections run in independent Tokio tasks and may process faults concurrently. Potentially @@ -373,9 +421,10 @@ accessor builds unless explicitly enabled. - The transport is local Unix stream plus `SCM_RIGHTS`; TCP is not supported. - Wire integers are little-endian. - Device and file offsets are byte offsets, not block numbers. -- `RANGE_RESPONSE` FD count must equal `range_count`. +- An `FD_RANGES` response's `fd_count`, payload entry count, and attached FD + count must match. - The client must keep its registered VMAs and userfaultfd alive for the connection lifetime. - Concurrent writers on one protocol connection are not currently supported; responses are serialized by the connection task. -- RANGE_RESPONSE batches belonging to one logical result are contiguous. +- `FD_RANGES` batches belonging to one logical result are contiguous. diff --git a/nydus/src/uffd/core.rs b/nydus/src/uffd/core.rs index 55102ecdf39..02510fe22cc 100644 --- a/nydus/src/uffd/core.rs +++ b/nydus/src/uffd/core.rs @@ -7,7 +7,7 @@ use anyhow::{anyhow, bail, Context, Result}; use crate::{Config, FdRange, NydusAccessor}; -use super::proto::{DeviceRange, FaultPolicy, VmaRegion}; +use super::proto::{DeviceRange, VmaRegion}; pub const UFFD_BLOCK_SIZE: u64 = 4096; pub const UFFD_TOTAL_SIZE_ALIGNMENT: u64 = 2 * 1024 * 1024; @@ -97,7 +97,7 @@ impl UffdCore { &self, uffd_fd: RawFd, regions: &[VmaRegion], - policy: FaultPolicy, + managed: bool, msg: &UffdMsg, ) -> Result> { let Some((region, range)) = resolve_fault_range(regions, msg)? else { @@ -105,20 +105,19 @@ impl UffdCore { }; let ranges = self.fetch_ranges(range.offset, range.len)?; - match policy { - FaultPolicy::Zerocopy => Ok(ranges), - FaultPolicy::Copy => { - for range in ranges { - let addr = region.virt_addr + (range.source_offset - region.offset); - if range.fd == self.accessor.zero_fd() { - uffdio_zeropage(uffd_fd, addr, range.len)?; - } else { - uffdio_copy_from_fd(uffd_fd, addr, range.fd, range.offset, range.len)?; - } - } - Ok(Vec::new()) + if !managed { + return Ok(ranges); + } + + for range in ranges { + let addr = region.virt_addr + (range.source_offset - region.offset); + if range.fd == self.accessor.zero_fd() { + uffdio_zeropage(uffd_fd, addr, range.len)?; + } else { + uffdio_copy_from_fd(uffd_fd, addr, range.fd, range.offset, range.len)?; } } + Ok(Vec::new()) } pub fn prefault_ranges(&self, regions: &[VmaRegion]) -> Result> { diff --git a/nydus/src/uffd/mod.rs b/nydus/src/uffd/mod.rs index 97510e05d5e..3dbb98f05cd 100644 --- a/nydus/src/uffd/mod.rs +++ b/nydus/src/uffd/mod.rs @@ -8,5 +8,5 @@ pub mod proto; pub mod service; pub use core::{UffdCore, UffdOptions}; -pub use proto::{FaultPolicy, VmaRegion}; +pub use proto::VmaRegion; pub use service::UffdService; diff --git a/nydus/src/uffd/proto.rs b/nydus/src/uffd/proto.rs index 8d6ee2c765b..c72eb7b35bf 100644 --- a/nydus/src/uffd/proto.rs +++ b/nydus/src/uffd/proto.rs @@ -1,6 +1,6 @@ -//! Nydus-compatible binary UFFD protocol definitions. +//! External userfaultfd backend protocol definitions. //! -//! Wire format: 20-byte little-endian header followed by typed payload. File +//! Wire format: 16-byte little-endian header followed by typed payload. File //! descriptors are passed with SCM_RIGHTS and are not counted in payload length. use std::io::{self, Write}; @@ -15,85 +15,118 @@ use sendfd::{RecvWithFd, SendWithFd}; use tokio::io::unix::AsyncFd; use tracing::warn; -pub const UFFD_MAGIC: u32 = 0x5546_4644; pub const UFFD_PROTOCOL_VERSION: u16 = 1; -pub const MSG_HANDSHAKE: u16 = 0x01; -pub const MSG_STAT_REQUEST: u16 = 0x02; -pub const MSG_FETCH_REQUEST: u16 = 0x03; -pub const MSG_PROBE_REQUEST: u16 = 0x04; +pub const COMMAND_HANDSHAKE: u16 = 0x0a; +pub const COMMAND_ADD_REGION: u16 = 0x0b; +pub const COMMAND_REMOVE_REGION: u16 = 0x0c; -pub const MSG_RANGE_RESPONSE: u16 = 0x81; -pub const MSG_STAT_RESPONSE: u16 = 0x82; +pub const COMMAND_STAT: u16 = 0x20; +pub const COMMAND_FETCH: u16 = 0x21; +pub const COMMAND_PROBE: u16 = 0x22; -pub const RANGE_RESPONSE_FLAG_NEXT: u16 = 1 << 0; +pub const STATUS_OK: u16 = 1; +pub const STATUS_ERROR: u16 = 2; -pub const HANDSHAKE_FLAG_COPY: u8 = 0x01; -pub const HANDSHAKE_FLAG_PREFAULT: u8 = 0x02; +pub const REPLY_LEGACY: u16 = 0; +pub const REPLY_FD_RANGES: u16 = 1; +pub const REPLY_STAT: u16 = 0x20; -pub const HEADER_SIZE: usize = 20; -pub const REGION_SIZE: usize = 40; +pub const FD_RANGES_FLAG_MORE: u16 = 1 << 0; + +pub const HANDSHAKE_FLAG_MANAGED: u8 = 1 << 0; +pub const HANDSHAKE_FLAG_PREFAULT: u8 = 1 << 1; +pub const HANDSHAKE_FLAG_ACK_REQUIRED: u8 = 1 << 2; +pub const HANDSHAKE_FLAG_BACKING_FDS: u8 = 1 << 3; + +pub const UFFD_MODE_MISSING: u8 = 1 << 0; +pub const UFFD_MODE_WP: u8 = 1 << 1; +pub const UFFD_MODE_WP_ASYNC: u8 = 1 << 2; + +pub const HEADER_SIZE: usize = 16; +pub const REGION_SIZE: usize = 48; pub const RANGE_SIZE: usize = 24; pub const FETCH_REQUEST_SIZE: usize = 16; pub const STAT_RESPONSE_SIZE: usize = size_of::() + 2 * size_of::(); -const HANDSHAKE_PREFIX_SIZE: usize = size_of::() + 2 * size_of::(); -const RANGE_COUNT_SIZE: usize = size_of::(); const MAX_RANGES_PER_MSG: usize = 16; -const MAX_RECV_FDS: usize = 32; +const MAX_RECV_FDS: usize = 64; const MAX_PAYLOAD_SIZE: usize = 64 * 1024; +pub(crate) type Frame = (RequestHeader, Vec, Vec); + #[derive(Debug, Clone, Copy)] -pub struct Header { - pub magic: u32, - pub flags: u16, - pub msg_type: u16, - pub cookie: u64, - pub len: u32, +pub struct RequestHeader { + pub command: u16, + pub command_headers: [u8; 6], + pub len: u64, } -impl Header { - pub fn new(msg_type: u16, payload_len: u32) -> Self { +impl RequestHeader { + pub fn new(command: u16, payload_len: u64) -> Self { Self { - magic: UFFD_MAGIC, - flags: 0, - msg_type, - cookie: 0, + command, + command_headers: [0; 6], len: payload_len, } } pub fn to_bytes(&self) -> [u8; HEADER_SIZE] { let mut buf = [0u8; HEADER_SIZE]; - buf[0..4].copy_from_slice(&self.magic.to_le_bytes()); - buf[4..6].copy_from_slice(&self.flags.to_le_bytes()); - buf[6..8].copy_from_slice(&self.msg_type.to_le_bytes()); - buf[8..16].copy_from_slice(&self.cookie.to_le_bytes()); - buf[16..20].copy_from_slice(&self.len.to_le_bytes()); + buf[0..2].copy_from_slice(&self.command.to_le_bytes()); + buf[2..8].copy_from_slice(&self.command_headers); + buf[8..16].copy_from_slice(&self.len.to_le_bytes()); buf } pub fn from_bytes(buf: &[u8; HEADER_SIZE]) -> Self { Self { - magic: u32::from_le_bytes(buf[0..4].try_into().unwrap()), - flags: u16::from_le_bytes(buf[4..6].try_into().unwrap()), - msg_type: u16::from_le_bytes(buf[6..8].try_into().unwrap()), - cookie: u64::from_le_bytes(buf[8..16].try_into().unwrap()), - len: u32::from_le_bytes(buf[16..20].try_into().unwrap()), + command: u16::from_le_bytes(buf[0..2].try_into().unwrap()), + command_headers: buf[2..8].try_into().unwrap(), + len: u64::from_le_bytes(buf[8..16].try_into().unwrap()), } } } -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -#[repr(u8)] -pub enum FaultPolicy { - #[default] - Zerocopy = 0, - Copy = 1, +#[derive(Debug, Clone, Copy)] +pub struct ResponseHeader { + pub status: u16, + pub reply_type: u16, + pub reply_headers: [u8; 4], + pub len: u64, +} + +impl ResponseHeader { + pub fn new(status: u16, reply_type: u16, payload_len: u64) -> Self { + Self { + status, + reply_type, + reply_headers: [0; 4], + len: payload_len, + } + } + + pub fn to_bytes(&self) -> [u8; HEADER_SIZE] { + let mut buf = [0u8; HEADER_SIZE]; + buf[0..2].copy_from_slice(&self.status.to_le_bytes()); + buf[2..4].copy_from_slice(&self.reply_type.to_le_bytes()); + buf[4..8].copy_from_slice(&self.reply_headers); + buf[8..16].copy_from_slice(&self.len.to_le_bytes()); + buf + } + + pub fn from_bytes(buf: &[u8; HEADER_SIZE]) -> Self { + Self { + status: u16::from_le_bytes(buf[0..2].try_into().unwrap()), + reply_type: u16::from_le_bytes(buf[2..4].try_into().unwrap()), + reply_headers: buf[4..8].try_into().unwrap(), + len: u64::from_le_bytes(buf[8..16].try_into().unwrap()), + } + } } #[repr(C)] -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct VmaRegion { pub virt_addr: u64, pub size: u64, @@ -101,6 +134,7 @@ pub struct VmaRegion { pub fault_size: u64, pub prot: i32, pub flags: i32, + pub backing_offset: u64, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -126,8 +160,7 @@ pub struct StatResponse { #[derive(Debug)] pub enum Request { Handshake { - policy: FaultPolicy, - prefault: bool, + flags: u8, regions: Vec, uffd: OwnedFd, }, @@ -155,34 +188,43 @@ impl ProtoConn { pub async fn recv(&self) -> Result> { loop { - let Some((msg_type, payload, mut fds)) = self.recv_frame().await? else { + let Some((header, payload, mut fds)) = self.recv_frame().await? else { return Ok(None); }; - let request = match msg_type { - MSG_HANDSHAKE => { - let (version, policy, prefault, regions) = decode_handshake(&payload) + let request = match header.command { + COMMAND_HANDSHAKE => { + let version = + u16::from_le_bytes(header.command_headers[0..2].try_into().unwrap()); + let flags = header.command_headers[2]; + let region_count = + u16::from_le_bytes(header.command_headers[4..6].try_into().unwrap()); + let regions = decode_handshake(region_count, &payload) .ok_or_else(|| anyhow!("invalid HANDSHAKE payload"))?; if version != UFFD_PROTOCOL_VERSION { bail!("unsupported UFFD protocol version {version}"); } - if fds.len() != 1 { + let expected_fds = 1 + if flags & HANDSHAKE_FLAG_BACKING_FDS != 0 { + usize::from(region_count) + } else { + 0 + }; + if fds.len() != expected_fds { bail!( - "HANDSHAKE must carry exactly one userfaultfd, received {}", + "HANDSHAKE must carry {expected_fds} file descriptor(s), received {}", fds.len() ); } Request::Handshake { - policy, - prefault, + flags, regions, - uffd: fds.pop().expect("validated HANDSHAKE fd count"), + uffd: fds.remove(0), } } - MSG_STAT_REQUEST => { + COMMAND_STAT => { validate_empty_request(&payload, &fds, "STAT")?; Request::Stat } - MSG_FETCH_REQUEST => { + COMMAND_FETCH => { validate_no_fds(&fds, "FETCH")?; let request = decode_fetch_request(&payload) .ok_or_else(|| anyhow!("invalid FETCH payload"))?; @@ -191,12 +233,12 @@ impl ProtoConn { len: request.len, }) } - MSG_PROBE_REQUEST => { + COMMAND_PROBE => { validate_empty_request(&payload, &fds, "PROBE")?; Request::Probe } other => { - warn!("nydus uffd ignored message type 0x{other:04x}"); + warn!("nydus uffd ignored command 0x{other:04x}"); continue; } }; @@ -204,7 +246,7 @@ impl ProtoConn { } } - async fn recv_frame(&self) -> Result, Vec)>> { + async fn recv_frame(&self) -> Result> { let mut header_buf = [0u8; HEADER_SIZE]; let mut raw_fds = [0i32; MAX_RECV_FDS]; let (read, fd_count) = recv_with_fd(&self.stream, &mut header_buf, &mut raw_fds).await?; @@ -219,10 +261,7 @@ impl ProtoConn { recv_exact(&self.stream, &mut header_buf[read..]).await?; } - let header = Header::from_bytes(&header_buf); - if header.magic != UFFD_MAGIC { - bail!("invalid UFFD magic 0x{:08x}", header.magic); - } + let header = RequestHeader::from_bytes(&header_buf); let payload_len = usize::try_from(header.len).context("invalid UFFD payload length")?; if payload_len > MAX_PAYLOAD_SIZE { bail!("UFFD payload length {payload_len} exceeds limit {MAX_PAYLOAD_SIZE}"); @@ -231,7 +270,7 @@ impl ProtoConn { if !payload.is_empty() { recv_exact(&self.stream, &mut payload).await?; } - Ok(Some((header.msg_type, payload, fds))) + Ok(Some((header, payload, fds))) } pub async fn send_ranges(&self, ranges: &[ResolvedRange]) -> Result<()> { @@ -256,6 +295,10 @@ impl ProtoConn { Ok(()) } + pub async fn send_ack(&self) -> Result<()> { + send_with_fd(&self.stream, &encode_ack_response(), &[]).await + } + pub async fn send_stat(&self, size: u64, block_size: u32, flags: u32) -> Result<()> { send_with_fd( &self.stream, @@ -266,57 +309,40 @@ impl ProtoConn { } } -pub fn encode_handshake( - ver: u16, - policy: FaultPolicy, - enable_prefault: bool, - regions: &[VmaRegion], -) -> Vec { - let payload_len = HANDSHAKE_PREFIX_SIZE + regions.len() * REGION_SIZE; - let header = Header::new(MSG_HANDSHAKE, payload_len as u32); - let mut flags = 0u8; - if policy == FaultPolicy::Copy { - flags |= HANDSHAKE_FLAG_COPY; - } - if enable_prefault { - flags |= HANDSHAKE_FLAG_PREFAULT; - } - +pub fn encode_handshake(ver: u16, flags: u8, uffd_modes: u8, regions: &[VmaRegion]) -> Vec { + let region_count = regions.len() as u16; + let payload_len = regions.len() * REGION_SIZE; + let mut header = RequestHeader::new(COMMAND_HANDSHAKE, payload_len as u64); + header.command_headers[0..2].copy_from_slice(&ver.to_le_bytes()); + header.command_headers[2] = flags; + header.command_headers[3] = uffd_modes; + header.command_headers[4..6].copy_from_slice(®ion_count.to_le_bytes()); let mut buf = Vec::with_capacity(HEADER_SIZE + payload_len); buf.extend_from_slice(&header.to_bytes()); - buf.extend_from_slice(&ver.to_le_bytes()); - buf.push(flags); - buf.push(regions.len() as u8); for r in regions { - buf.extend_from_slice(&r.virt_addr.to_le_bytes()); - buf.extend_from_slice(&r.size.to_le_bytes()); - buf.extend_from_slice(&r.offset.to_le_bytes()); - buf.extend_from_slice(&r.fault_size.to_le_bytes()); - buf.extend_from_slice(&r.prot.to_le_bytes()); - buf.extend_from_slice(&r.flags.to_le_bytes()); + encode_region(&mut buf, r); } buf } -pub fn decode_handshake(payload: &[u8]) -> Option<(u16, FaultPolicy, bool, Vec)> { - if payload.len() < HANDSHAKE_PREFIX_SIZE { - return None; - } - let ver = u16::from_le_bytes(payload[0..2].try_into().unwrap()); - let flags = payload[2]; - let region_count = payload[3] as usize; - let expected_len = HANDSHAKE_PREFIX_SIZE + region_count * REGION_SIZE; +fn encode_region(buf: &mut Vec, region: &VmaRegion) { + buf.extend_from_slice(®ion.virt_addr.to_le_bytes()); + buf.extend_from_slice(®ion.size.to_le_bytes()); + buf.extend_from_slice(®ion.offset.to_le_bytes()); + buf.extend_from_slice(®ion.fault_size.to_le_bytes()); + buf.extend_from_slice(®ion.prot.to_le_bytes()); + buf.extend_from_slice(®ion.flags.to_le_bytes()); + buf.extend_from_slice(®ion.backing_offset.to_le_bytes()); +} + +pub fn decode_handshake(region_count: u16, payload: &[u8]) -> Option> { + let region_count = region_count as usize; + let expected_len = region_count * REGION_SIZE; if payload.len() != expected_len { return None; } - let policy = if flags & HANDSHAKE_FLAG_COPY != 0 { - FaultPolicy::Copy - } else { - FaultPolicy::Zerocopy - }; - let enable_prefault = flags & HANDSHAKE_FLAG_PREFAULT != 0; let mut regions = Vec::with_capacity(region_count); - let mut off = HANDSHAKE_PREFIX_SIZE; + let mut off = 0; for _ in 0..region_count { regions.push(VmaRegion { virt_addr: u64::from_le_bytes(payload[off..off + 8].try_into().unwrap()), @@ -325,21 +351,28 @@ pub fn decode_handshake(payload: &[u8]) -> Option<(u16, FaultPolicy, bool, Vec Vec { - let payload_len = RANGE_COUNT_SIZE + ranges.len() * RANGE_SIZE; - let mut header = Header::new(MSG_RANGE_RESPONSE, payload_len as u32); - if next { - header.flags |= RANGE_RESPONSE_FLAG_NEXT; - } +pub fn encode_ack_response() -> Vec { + ResponseHeader::new(STATUS_OK, REPLY_LEGACY, 0) + .to_bytes() + .to_vec() +} + +pub fn encode_range_response(ranges: &[(u64, u64, u64)], more: bool) -> Vec { + let payload_len = ranges.len() * RANGE_SIZE; + let mut header = ResponseHeader::new(STATUS_OK, REPLY_FD_RANGES, payload_len as u64); + let flags = if more { FD_RANGES_FLAG_MORE } else { 0 }; + let fd_count = ranges.len() as u16; + header.reply_headers[0..2].copy_from_slice(&flags.to_le_bytes()); + header.reply_headers[2..4].copy_from_slice(&fd_count.to_le_bytes()); let mut buf = Vec::with_capacity(HEADER_SIZE + payload_len); buf.extend_from_slice(&header.to_bytes()); - buf.extend_from_slice(&(ranges.len() as u32).to_le_bytes()); for &(device_offset, blob_offset, len) in ranges { buf.extend_from_slice(&device_offset.to_le_bytes()); buf.extend_from_slice(&blob_offset.to_le_bytes()); @@ -349,33 +382,26 @@ pub fn encode_range_response(ranges: &[(u64, u64, u64)], next: bool) -> Vec } pub fn decode_range_response(payload: &[u8]) -> Option> { - if payload.len() < RANGE_COUNT_SIZE { - return None; - } - let range_count = u32::from_le_bytes(payload[..RANGE_COUNT_SIZE].try_into().unwrap()) as usize; - let expected_len = RANGE_COUNT_SIZE + range_count * RANGE_SIZE; - if payload.len() != expected_len { + if !payload.len().is_multiple_of(RANGE_SIZE) { return None; } - let mut ranges = Vec::with_capacity(range_count); - let mut off = RANGE_COUNT_SIZE; - for _ in 0..range_count { + let mut ranges = Vec::with_capacity(payload.len() / RANGE_SIZE); + for entry in payload.chunks_exact(RANGE_SIZE) { ranges.push(BlobRange { - device_offset: u64::from_le_bytes(payload[off..off + 8].try_into().unwrap()), - blob_offset: u64::from_le_bytes(payload[off + 8..off + 16].try_into().unwrap()), - len: u64::from_le_bytes(payload[off + 16..off + 24].try_into().unwrap()), + device_offset: u64::from_le_bytes(entry[0..8].try_into().unwrap()), + blob_offset: u64::from_le_bytes(entry[8..16].try_into().unwrap()), + len: u64::from_le_bytes(entry[16..24].try_into().unwrap()), }); - off += RANGE_SIZE; } Some(ranges) } pub fn encode_stat_request() -> Vec { - Header::new(MSG_STAT_REQUEST, 0).to_bytes().to_vec() + RequestHeader::new(COMMAND_STAT, 0).to_bytes().to_vec() } pub fn encode_stat_response(size: u64, block_size: u32, flags: u32) -> Vec { - let header = Header::new(MSG_STAT_RESPONSE, STAT_RESPONSE_SIZE as u32); + let header = ResponseHeader::new(STATUS_OK, REPLY_STAT, STAT_RESPONSE_SIZE as u64); let mut buf = Vec::with_capacity(HEADER_SIZE + STAT_RESPONSE_SIZE); buf.extend_from_slice(&header.to_bytes()); buf.extend_from_slice(&size.to_le_bytes()); @@ -396,7 +422,11 @@ pub fn decode_stat_response(payload: &[u8]) -> Option { } pub fn encode_fetch_request(offset: u64, len: u64) -> Vec { - let header = Header::new(MSG_FETCH_REQUEST, FETCH_REQUEST_SIZE as u32); + encode_range_request(COMMAND_FETCH, offset, len) +} + +fn encode_range_request(command: u16, offset: u64, len: u64) -> Vec { + let header = RequestHeader::new(command, FETCH_REQUEST_SIZE as u64); let mut buf = Vec::with_capacity(HEADER_SIZE + FETCH_REQUEST_SIZE); buf.extend_from_slice(&header.to_bytes()); buf.extend_from_slice(&offset.to_le_bytes()); @@ -415,7 +445,7 @@ pub fn decode_fetch_request(payload: &[u8]) -> Option { } pub fn encode_probe_request() -> Vec { - Header::new(MSG_PROBE_REQUEST, 0).to_bytes().to_vec() + RequestHeader::new(COMMAND_PROBE, 0).to_bytes().to_vec() } fn validate_no_fds(fds: &[OwnedFd], name: &str) -> Result<()> { @@ -518,12 +548,54 @@ mod tests { (ProtoConn::new(server).unwrap(), client) } + #[test] + fn protocol_number_layout() { + assert_eq!(COMMAND_HANDSHAKE, 0x0a); + assert_eq!(COMMAND_ADD_REGION, 0x0b); + assert_eq!(COMMAND_REMOVE_REGION, 0x0c); + assert_eq!(COMMAND_STAT, 0x20); + assert_eq!(COMMAND_FETCH, 0x21); + assert_eq!(COMMAND_PROBE, 0x22); + + assert_eq!(REPLY_LEGACY, 0); + assert_eq!(REPLY_FD_RANGES, 1); + assert_eq!(REPLY_STAT, 0x20); + } + + #[test] + fn ack_response_wire_layout() { + let buf = encode_ack_response(); + let header = ResponseHeader::from_bytes(&buf.try_into().unwrap()); + assert_eq!(header.status, STATUS_OK); + assert_eq!(header.reply_type, REPLY_LEGACY); + assert_eq!(header.reply_headers, [0; 4]); + assert_eq!(header.len, 0); + } + + #[test] + fn range_response_roundtrip() { + let buf = encode_range_response(&[(0, 4096, 8192)], false); + let hdr = ResponseHeader::from_bytes(&buf[..HEADER_SIZE].try_into().unwrap()); + assert_eq!(hdr.status, STATUS_OK); + assert_eq!(hdr.reply_type, REPLY_FD_RANGES); + let flags = u16::from_le_bytes(hdr.reply_headers[0..2].try_into().unwrap()); + let fd_count = u16::from_le_bytes(hdr.reply_headers[2..4].try_into().unwrap()); + assert_eq!(flags, 0); + assert_eq!(fd_count, 1); + assert_eq!(hdr.len as usize, RANGE_SIZE); + let ranges = decode_range_response(&buf[HEADER_SIZE..]).unwrap(); + assert_eq!(ranges[0].device_offset, 0); + assert_eq!(ranges[0].blob_offset, 4096); + assert_eq!(ranges[0].len, 8192); + } + #[test] fn stat_response_roundtrip() { let buf = encode_stat_response(0x20_0000, 4096, 1); - let hdr = Header::from_bytes(&buf[..HEADER_SIZE].try_into().unwrap()); - assert_eq!(hdr.msg_type, MSG_STAT_RESPONSE); - assert_eq!(MSG_STAT_RESPONSE, 0x82); + let hdr = ResponseHeader::from_bytes(&buf[..HEADER_SIZE].try_into().unwrap()); + assert_eq!(hdr.status, STATUS_OK); + assert_eq!(hdr.reply_type, REPLY_STAT); + assert_eq!(hdr.reply_headers, [0; 4]); assert_eq!(hdr.len as usize, STAT_RESPONSE_SIZE); assert_eq!( decode_stat_response(&buf[HEADER_SIZE..]), @@ -535,24 +607,12 @@ mod tests { ); } - #[test] - fn range_response_roundtrip() { - let buf = encode_range_response(&[(0, 4096, 8192)], false); - let hdr = Header::from_bytes(&buf[..HEADER_SIZE].try_into().unwrap()); - assert_eq!(hdr.magic, UFFD_MAGIC); - assert_eq!(hdr.msg_type, MSG_RANGE_RESPONSE); - assert_eq!(hdr.flags, 0); - let ranges = decode_range_response(&buf[HEADER_SIZE..]).unwrap(); - assert_eq!(ranges[0].device_offset, 0); - assert_eq!(ranges[0].blob_offset, 4096); - assert_eq!(ranges[0].len, 8192); - } - #[test] fn fetch_request_roundtrip() { let buf = encode_fetch_request(0x1234_0000, 0x20_0000); - let hdr = Header::from_bytes(&buf[..HEADER_SIZE].try_into().unwrap()); - assert_eq!(hdr.msg_type, MSG_FETCH_REQUEST); + let hdr = RequestHeader::from_bytes(&buf[..HEADER_SIZE].try_into().unwrap()); + assert_eq!(hdr.command, COMMAND_FETCH); + assert_eq!(hdr.command_headers, [0; 6]); assert_eq!(hdr.len as usize, FETCH_REQUEST_SIZE); assert_eq!( decode_fetch_request(&buf[HEADER_SIZE..]), @@ -567,10 +627,7 @@ mod tests { #[tokio::test] async fn proto_conn_decodes_typed_requests() { let (proto, client) = proto_pair(); - let mut stat = Header::from_bytes(&encode_stat_request().try_into().unwrap()); - stat.flags = 1; - stat.cookie = 2; - client.send_with_fd(&stat.to_bytes(), &[]).unwrap(); + client.send_with_fd(&encode_stat_request(), &[]).unwrap(); assert!(matches!(proto.recv().await.unwrap(), Some(Request::Stat))); client @@ -599,43 +656,65 @@ mod tests { fault_size: 0x1000, prot: 1, flags: 2, + backing_offset: 0x4000, }; + let backing = File::open("/dev/zero").unwrap(); + let flags = HANDSHAKE_FLAG_MANAGED | HANDSHAKE_FLAG_PREFAULT | HANDSHAKE_FLAG_ACK_REQUIRED; + let mut handshake = encode_handshake( + UFFD_PROTOCOL_VERSION, + flags, + UFFD_MODE_MISSING, + std::slice::from_ref(®ion), + ); + handshake[4] |= HANDSHAKE_FLAG_BACKING_FDS | (1 << 7); + handshake[5] |= 1 << 7; + let header = RequestHeader::from_bytes(&handshake[..HEADER_SIZE].try_into().unwrap()); + assert_eq!(header.command, COMMAND_HANDSHAKE); + assert_eq!( + header.command_headers[2] & HANDSHAKE_FLAG_MANAGED, + HANDSHAKE_FLAG_MANAGED + ); + assert_eq!( + header.command_headers[2] & HANDSHAKE_FLAG_ACK_REQUIRED, + HANDSHAKE_FLAG_ACK_REQUIRED + ); + assert_eq!( + header.command_headers[3] & UFFD_MODE_MISSING, + UFFD_MODE_MISSING + ); + assert_eq!(header.len as usize, REGION_SIZE); client - .send_with_fd( - &encode_handshake( - UFFD_PROTOCOL_VERSION, - FaultPolicy::Copy, - true, - std::slice::from_ref(®ion), - ), - &[file.as_raw_fd()], - ) + .send_with_fd(&handshake, &[file.as_raw_fd(), backing.as_raw_fd()]) .unwrap(); drop(file); + drop(backing); let Some(Request::Handshake { - policy, - prefault, + flags, regions, uffd, }) = proto.recv().await.unwrap() else { panic!("expected HANDSHAKE request"); }; - assert_eq!(policy, FaultPolicy::Copy); - assert!(prefault); + assert_eq!(flags & HANDSHAKE_FLAG_MANAGED, HANDSHAKE_FLAG_MANAGED); + assert_eq!(flags & HANDSHAKE_FLAG_PREFAULT, HANDSHAKE_FLAG_PREFAULT); + assert_eq!( + flags & HANDSHAKE_FLAG_ACK_REQUIRED, + HANDSHAKE_FLAG_ACK_REQUIRED + ); assert_eq!(regions, vec![region]); assert!(unsafe { libc::fcntl(uffd.as_raw_fd(), libc::F_GETFD) } >= 0); } #[tokio::test] async fn proto_conn_rejects_invalid_handshake_fd_counts() { - let handshake = encode_handshake(UFFD_PROTOCOL_VERSION, FaultPolicy::Zerocopy, false, &[]); + let handshake = encode_handshake(UFFD_PROTOCOL_VERSION, 0, UFFD_MODE_MISSING, &[]); let (proto, client) = proto_pair(); client.send_with_fd(&handshake, &[]).unwrap(); let err = proto.recv().await.unwrap_err(); - assert!(format!("{err:#}").contains("exactly one userfaultfd, received 0")); + assert!(format!("{err:#}").contains("must carry 1 file descriptor(s), received 0")); let (proto, client) = proto_pair(); let first = File::open("/dev/null").unwrap(); @@ -644,7 +723,7 @@ mod tests { .send_with_fd(&handshake, &[first.as_raw_fd(), second.as_raw_fd()]) .unwrap(); let err = proto.recv().await.unwrap_err(); - assert!(format!("{err:#}").contains("exactly one userfaultfd, received 2")); + assert!(format!("{err:#}").contains("must carry 1 file descriptor(s), received 2")); } #[tokio::test] @@ -660,8 +739,8 @@ mod tests { let (read, fd_count) = client.recv_with_fd(&mut header_buf, &mut raw_fds).unwrap(); assert_eq!(read, HEADER_SIZE); assert_eq!(fd_count, 0); - let header = Header::from_bytes(&header_buf); - assert_eq!(header.msg_type, MSG_STAT_RESPONSE); + let header = ResponseHeader::from_bytes(&header_buf); + assert_eq!(header.reply_type, REPLY_STAT); let mut payload = vec![0u8; header.len as usize]; client.read_exact(&mut payload).unwrap(); assert_eq!(decode_stat_response(&payload).unwrap().size, 0x20_0000); @@ -676,7 +755,7 @@ mod tests { #[tokio::test] async fn proto_conn_rejects_oversized_payload() { let (proto, client) = proto_pair(); - let header = Header::new(MSG_FETCH_REQUEST, (MAX_PAYLOAD_SIZE + 1) as u32); + let header = RequestHeader::new(COMMAND_FETCH, (MAX_PAYLOAD_SIZE + 1) as u64); client.send_with_fd(&header.to_bytes(), &[]).unwrap(); let err = proto.recv().await.unwrap_err(); assert!(format!("{err:#}").contains("exceeds limit")); @@ -707,7 +786,7 @@ mod tests { .collect::>(); proto.send_ranges(&ranges).await.unwrap(); - for (expected_count, expected_flags) in [(16, RANGE_RESPONSE_FLAG_NEXT), (1, 0)] { + for (expected_count, expected_flags) in [(16, FD_RANGES_FLAG_MORE), (1, 0)] { let mut header_buf = [0u8; HEADER_SIZE]; let mut raw_fds = [0i32; MAX_RECV_FDS]; let (read, fd_count) = client.recv_with_fd(&mut header_buf, &mut raw_fds).unwrap(); @@ -717,9 +796,12 @@ mod tests { .iter() .map(|fd| unsafe { OwnedFd::from_raw_fd(*fd) }) .collect::>(); - let header = Header::from_bytes(&header_buf); - assert_eq!(header.msg_type, MSG_RANGE_RESPONSE); - assert_eq!(header.flags, expected_flags); + let header = ResponseHeader::from_bytes(&header_buf); + assert_eq!(header.reply_type, REPLY_FD_RANGES); + let flags = u16::from_le_bytes(header.reply_headers[0..2].try_into().unwrap()); + let fd_count = u16::from_le_bytes(header.reply_headers[2..4].try_into().unwrap()); + assert_eq!(flags, expected_flags); + assert_eq!(fd_count as usize, expected_count); let mut payload = vec![0u8; header.len as usize]; client.read_exact(&mut payload).unwrap(); assert_eq!( @@ -740,9 +822,12 @@ mod tests { let (read, fd_count) = client.recv_with_fd(&mut header_buf, &mut raw_fds).unwrap(); assert_eq!(read, HEADER_SIZE); assert_eq!(fd_count, 0); - let header = Header::from_bytes(&header_buf); - assert_eq!(header.msg_type, MSG_RANGE_RESPONSE); - assert_eq!(header.flags, 0); + let header = ResponseHeader::from_bytes(&header_buf); + assert_eq!(header.reply_type, REPLY_FD_RANGES); + let flags = u16::from_le_bytes(header.reply_headers[0..2].try_into().unwrap()); + let fd_count = u16::from_le_bytes(header.reply_headers[2..4].try_into().unwrap()); + assert_eq!(flags, 0); + assert_eq!(fd_count, 0); let mut payload = vec![0u8; header.len as usize]; client.read_exact(&mut payload).unwrap(); assert!(decode_range_response(&payload).unwrap().is_empty()); diff --git a/nydus/src/uffd/service.rs b/nydus/src/uffd/service.rs index b3717b6cdb6..4f5ff5c2546 100644 --- a/nydus/src/uffd/service.rs +++ b/nydus/src/uffd/service.rs @@ -15,7 +15,10 @@ use tracing::{debug, info, warn}; use crate::FdRange; use super::core::{read_uffd_msg, UffdCore, UffdMsg}; -use super::proto::{FaultPolicy, ProtoConn, Request, VmaRegion}; +use super::proto::{ + ProtoConn, Request, VmaRegion, HANDSHAKE_FLAG_ACK_REQUIRED, HANDSHAKE_FLAG_MANAGED, + HANDSHAKE_FLAG_PREFAULT, +}; pub struct UffdService { core: Arc, @@ -25,7 +28,7 @@ pub struct UffdService { struct HandshakeState { regions: Vec, - policy: FaultPolicy, + managed: bool, uffd: AsyncFd, } @@ -205,18 +208,14 @@ impl UffdConn { }; match request { Request::Handshake { - policy, - prefault, + flags, regions, uffd, } => { if self.state.is_some() { bail!("duplicate UFFD handshake"); } - self.state = Some( - self.handle_handshake(policy, prefault, regions, uffd) - .await?, - ); + self.state = Some(self.handle_handshake(flags, regions, uffd).await?); } Request::Stat => { self.proto @@ -242,18 +241,20 @@ impl UffdConn { async fn handle_handshake( &self, - policy: FaultPolicy, - prefault: bool, + flags: u8, regions: Vec, uffd: OwnedFd, ) -> Result { + let managed = flags & HANDSHAKE_FLAG_MANAGED != 0; + let prefault = flags & HANDSHAKE_FLAG_PREFAULT != 0; + let ack_required = flags & HANDSHAKE_FLAG_ACK_REQUIRED != 0; set_nonblocking(uffd.as_raw_fd())?; let async_uffd = AsyncFd::new(uffd).context("failed to register userfaultfd with tokio")?; info!( - "nydus uffd handshake: regions={} policy={:?} prefault={}", + "nydus uffd handshake: regions={} managed={} prefault={}", regions.len(), - policy, + managed, prefault ); for (idx, region) in regions.iter().enumerate() { @@ -268,15 +269,19 @@ impl UffdConn { ); } + if ack_required { + self.proto.send_ack().await?; + } + // Keep prefault synchronous until ProtoConn guarantees serialized concurrent writes. - if prefault && policy == FaultPolicy::Zerocopy { + if prefault && !managed { let ranges = self.core.prefault_ranges(®ions)?; self.send_ranges(Some(®ions), &ranges).await?; } Ok(HandshakeState { regions, - policy, + managed, uffd: async_uffd, }) } @@ -287,7 +292,7 @@ impl UffdConn { .as_ref() .ok_or_else(|| anyhow!("received UFFD event before handshake"))?; let uffd_fd = state.uffd.get_ref().as_raw_fd(); - let policy = state.policy; + let managed = state.managed; let regions = state.regions.clone(); debug!( "nydus uffd event: event=0x{:02x} addr={:#x} flags={:#x}", @@ -295,17 +300,17 @@ impl UffdConn { ); let core = self.core.clone(); let ranges = tokio::task::spawn_blocking(move || { - core.resolve_page_fault(uffd_fd, ®ions, policy, &msg) + core.resolve_page_fault(uffd_fd, ®ions, managed, &msg) }) .await .context("UFFD page-fault blocking task failed")??; debug!( - "nydus uffd resolved fault: policy={:?} ranges={}", - policy, + "nydus uffd resolved fault: managed={} ranges={}", + managed, ranges.len() ); - if policy == FaultPolicy::Zerocopy { + if !managed { self.send_ranges(Some(&state.regions), &ranges).await?; } Ok(())