diff --git a/AGENTS.md b/AGENTS.md index 716006b..343b5c0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -2,7 +2,7 @@ ## Overview -- This is a test project meant to bypass/replace VDDK, which is no longer +- This is a project meant to bypass/replace VDDK, which is no longer publicly available. - The end goal is to have a Python library that can be used as a VDDK replacement to retrieve VMware disk contents. diff --git a/README.md b/README.md index 7fab225..254794b 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,10 @@ OpenVixDiskLib is an open-source Python replacement for VMware VDDK's `VixDiskLib` NBD path. It reads and writes VMDK contents over vSphere NFC without the proprietary VDDK SDK. +AI tools (Cursor + Grok 4.6) have been heavily used to reverse engineer the +NBD and NFC protocols, obtaining a working VDDK replacement in a few hours and +comprehensive testing in a matter of days. + The Python package is `openvixdisklib` (lowercase, following usual Python naming). @@ -18,7 +22,7 @@ Implemented against vCenter 8 / ESXi 8. Default transport is `nbdssl` - `VixDiskLib_ConnectEx` (UID credentials) - `VixDiskLib_Open` (datastore path, read-only or read-write) -- `VixDiskLib_Read` +- `VixDiskLib_Read` (optional ``skip_decompression`` packs FastLZ extras) - `VixDiskLib_Write` Not implemented: compression open flags other than FastLZ, CBT / @@ -112,7 +116,8 @@ tox -e integration -- --runslow Compare write/read throughput of OpenVixDiskLib and native VDDK (`64KiB`, 129-sector, and `32MiB` transfers; `nbdssl` and `nbd`; -plain and FastLZ): +plain, FastLZ, and OpenVixDiskLib FastLZ ``skip_decompression``; +AIO sessions 64 KiB×1, 1 MiB×1, 2 MiB×1, and 2 MiB×4). ```bash tox -e perf diff --git a/docs/nfc_open.md b/docs/nfc_open.md index 6a72962..bf61550 100644 --- a/docs/nfc_open.md +++ b/docs/nfc_open.md @@ -158,8 +158,26 @@ obtain a file handle or to read sector 0. ### OPEN_SESSION / sockopts / resource pool -VDDK sends 16 zero bytes (`OPEN_SESSION`; server replies with 16 zeros), -12 zero bytes (`SET_SOCK_OPTS`; server returns send/recv buffer sizes +`OPEN_SESSION` payload is 16 bytes, little-endian: + +| Offset | Type | Meaning | +| ------ | -------- | ---------------------------------------------------- | +| 0 | `uint32` | 0 (unused in captures) | +| 4 | `uint32` | AIO buffer size in **bytes** (VDDK default 65536) | +| 8 | `uint32` | Buffer count (VDDK ``nfcAio.Session.BufCount``) | +| 12 | `uint32` | 0 | + +VDDK config `vixDiskLib.nfcAio.Session.BufSizeIn64KB` is that byte size +divided by 64 KiB (`1` → 65536, `32` → 2097152). The server replies +with 16 zeros; it still **uses** the requested size for IO extras. +A 129-sector read is two fragments at 64 KiB, and one 66048-byte +fragment at 2 MiB. Lab ESXi 8 accepted 2 MiB (`BufCount` 1 and 4) and +rejected 16 MiB and 32 MiB (`OPEN_SESSION` AIO error). Broadcom's 16 MiB +figure is session memory (`size × count`), not a larger extra; the +per-buffer max on the wire is 2 MiB. Probe: +`docs/probing_samples/vddk_aio_bufsize_probe.py`. + +`SET_SOCK_OPTS` is 12 zero bytes (server returns send/recv buffer sizes and a `uint32` flag), then `uint32` 1 (`SET_RES_POOL`, log: “Setting Resource Pool(1)”). @@ -205,14 +223,15 @@ classic type 4 `NFC_SESSION_COMPLETE`. ## OpenVixDiskLib -| Piece | Module | -| ----------------------------- | ----------------------------------------------- | -| VIM + authd | `openvixdisklib.nfc_auth.authenticate` | -| Dup fd, skip TLS for NFC | `openvixdisklib.nfc_open.takeover_authd_socket` | -| Second TLS for nbdssl | `openvixdisklib.nfc_open.wrap_nfcssl_socket` | -| FastLZ for NBD compression | `openvixdisklib.fastlz` (pip `pyfastlz`) | -| Handshake + AIO + OPEN_FILE | `openvixdisklib.nfc_open.open_disk` | -| Sector read / write / close | `openvixdisklib.nfc_open.NfcDisk` | +| Piece | Module | +| ------------------------------- | --------------------------------------------------------- | +| VIM + authd | `openvixdisklib.nfc_auth.authenticate` | +| Dup fd, skip TLS for NFC | `openvixdisklib.nfc_open.takeover_authd_socket` | +| Second TLS for nbdssl | `openvixdisklib.nfc_open.wrap_nfcssl_socket` | +| FastLZ for NBD compression | `openvixdisklib.fastlz` (pip `pyfastlz`) | +| Handshake + AIO + OPEN_FILE | `openvixdisklib.nfc_open.open_disk` | +| AIO extra size / pool count | `open_disk(..., aio_buffer_size=, aio_buffer_count=)` | +| Sector read / write / close | `openvixdisklib.nfc_open.NfcDisk` | Run: diff --git a/docs/nfc_read.md b/docs/nfc_read.md index 2afa3c2..ee67652 100644 --- a/docs/nfc_read.md +++ b/docs/nfc_read.md @@ -26,11 +26,13 @@ length = numSectors * sectorSize | `VixDiskLib_Read(h, 0, 128, buf)` | IO length 65536 (AIO buffer size), one fragment | | `VixDiskLib_Read(h, 0, 129, buf)` | One request of 66048; **two** reply fragments | -VDDK does **not** split a `Read` larger than 64 KiB into multiple -requests. The client sends one AIO message; the server answers with -one or more same-`opId` replies, each carrying at most -`NFC_AIO_BUFFER_SIZE` (65536) data bytes. `NfcAioInitSession` logged -that buffer size and count 4 during open. +VDDK does **not** split a `Read` larger than the AIO buffer into +multiple requests. The client sends one AIO message; the server +answers with one or more same-`opId` replies, each carrying at most +the OPEN_SESSION buffer size (VDDK default 65536). +`vixDiskLib.nfcAio.Session.BufSizeIn64KB=32` advertises 2 MiB; a +129-sector read then returns **one** 66048-byte extra, and a 2 MiB + +512 read returns 2097152 + 512. See `docs/nfc_open.md` (OPEN_SESSION). Sparse regions are still transferred as zeros. A read of 8 sectors at LBA 8 on this disk was 4096 zero bytes on the wire, not a skip. @@ -81,22 +83,23 @@ payload + `chunkLength` data bytes. Reply payload (handle is zeroed; lengths describe this fragment): -| Offset | Type | Meaning | -| ------ | -------- | ----------------------------------------------- | -| 0 | `uint64` | `0` | -| 8 | `uint64` | `1` (read) | -| 16 | `uint64` | Byte offset of the **request** | -| 24 | `uint32` | Total request length | -| 28 | `uint32` | Byte offset of this fragment (`0`, `65536`, …) | -| 32 | `uint32` | This fragment’s byte length | -| 36 | `uint32` | Same as offset 32 | -| 40 | `uint32` | `0` | +| Offset | Type | Meaning | +| ------ | -------- | -------------------------------------------------------------------- | +| 0 | `uint64` | `0` | +| 8 | `uint64` | `1` (read) | +| 16 | `uint64` | Byte offset of the **request** on disk | +| 24 | `uint32` | Total request length | +| 28 | `uint32` | Fragment byte offset **in this request** (`0`, `65536`, …), not disk | +| 32 | `uint32` | This fragment’s uncompressed byte length | +| 36 | `uint32` | Same as offset 32, or compressed extra size when type is FastLZ | +| 40 | `uint32` | `0` | When there is a single fragment, offsets 24–31 look like a `uint64` length (the fragment offset is 0). The 129-sector capture shows why they are two `uint32`s: fragment 0 has `(66048, 0)` then chunk 65536; fragment 1 has `(66048, 65536)` then chunk 512. `0x00010000` at offset -28 is the byte offset, not a 0-based index. +28 is the byte offset, not a 0-based index. Disk byte address of a +fragment is request offset (payload 16) plus payload 28. Read loop: receive fragments with that `opId` until the concatenated data length equals the request. Use the `uint32` at payload offset 32 @@ -113,6 +116,9 @@ S: type=7 opId=18 size=44 dest=0 chunk=65536 + 65536 data S: type=7 opId=18 size=44 dest=65536 chunk=512 + 512 data ``` +`dest` in that dump is payload offset 28 (`ReadFragment.dest`): 0 and +65536 are positions in this 66048-byte read, not sector numbers. + ## Lab check Integration tests create an empty 10 GiB thin disk, write a repeating @@ -135,6 +141,54 @@ The integration test writes and then reads the captured VDDK ranges (including a 129-sector transfer that must assemble two read fragments). +## Skip decompression (OpenVixDiskLib extension) + +`VixDiskLib_Read` always fills `buf` with uncompressed sector bytes. +OpenVixDiskLib can skip FastLZ decode so a backup application can +forward the compressed data as-is, avoiding unnecessary re-compression. + +`NfcDisk.readinto(..., skip_decompression=True)` and +`VixDiskLibHandle.read(..., skip_decompression=True)` still send one +IO request and wait until uncompressed `filled == length`. They do +**not** decompress. Extras are packed densely from offset 0 of `buf`. +`ReadResult.fragments` describes each extra. Type `2` extras are +FastLZ; type `0` fallbacks are raw. Concatenating extras is not a +valid FastLZ stream; the caller must use the table to split them. + +| Field | Meaning | +| ---------------------- | ------------------------------------------------------------------------------------------------ | +| `dest` | Byte offset **in this uncompressed read** (NFC payload 28). Not a disk LBA or VMDK file offset. | +| `uncompressed_length` | Uncompressed fragment size (NFC payload 32). | +| `compression_type` | `NFC_COMPRESSION_NONE` (0) or `NFC_COMPRESSION_FASTLZ` (2). | +| `offset` | Start of this extra in packed `buf` (receive order, densely from 0). | +| `length` | Extra size on the wire. | + +Disk byte address of a fragment is `start_sector * 512 + dest`. A +129-sector `read` from sector 0 or from sector 1000 still reports +`dest=0` and `dest=65536` when extras are 64 KiB. + +`buf` is sized for the uncompressed request, so it is always large +enough. Default `read` still decompresses; `fragments` is empty and +`compressed_length` is still the extra bytes on the wire. +`skip_decompression` with a plain (no FASTLZ) open only records raw +extras (`compressed_length == uncompressed_length`). + +This is not `VixDiskLib_Read`. Do not add an open flag for it; +compression on the wire is already the FASTLZ open flag. + +A 32 MiB read at 64 KiB extras is 512 fragments in **one** result. A +2 MiB OPEN_SESSION extra (`aio_buffer_size=2097152`) is 16 fragments +for the same read. One dest PUT per extra is not viable. + +``` +uncompressed request (offsets in this read, not on disk) +|---------------- 64KiB --|-- 64KiB --|-- ... --| + dest=0 dest=65536 + extra (FastLZ or raw) extra (FastLZ or raw) + +buf when skip_decompression=True: extras packed densely from offset 0 +``` + ## What is still VDDK-only - zlib and skipz NBD compression flags diff --git a/docs/nfc_write.md b/docs/nfc_write.md index 79993ca..a2c59d6 100644 --- a/docs/nfc_write.md +++ b/docs/nfc_write.md @@ -85,8 +85,9 @@ A 1-sector VDDK write was 572 bytes on the wire: 16 + 44 + 512. ## Fragments and the single reply -`NfcAioInitSession` advertises a 64 KiB buffer. Extra per type-7 -message is at most that size. VDDK does **not** issue a new `opId` per +`OPEN_SESSION` advertises the AIO buffer size (VDDK default 64 KiB; +`BufSizeIn64KB` can raise it). Extra per type-7 message is at most +that size. VDDK does **not** issue a new `opId` per chunk, and it does **not** coalesce separate `VixDiskLib_Write` calls (eight 8 KiB writes stayed eight IOs). One public write becomes N client type-7 messages with the **same** `opId`, then **one** 44-byte @@ -105,9 +106,9 @@ for one reply per chunk; raising the window did not match VDDK throughput because VDDK pays one RTT per `Write`, not per fragment. `NfcAioFlushCoalescedWrites` is server-side (`nfcAioServer.c`), not a -client merge of API writes. OPEN_SESSION is 16 zero bytes both ways, so -the logged AIO buffer count of 4 is a VDDK client default -(`vixDiskLib.nfcAio.Session.BufCount`), not a server cap. +client merge of API writes. Buffer count is +`vixDiskLib.nfcAio.Session.BufCount` (OPEN_SESSION offset 8); size is +`BufSizeIn64KB` (offset 4, in bytes). `docs/nfc_open.md`. ## OpenVixDiskLib diff --git a/docs/probing_samples/vddk_aio_bufsize_probe.py b/docs/probing_samples/vddk_aio_bufsize_probe.py new file mode 100644 index 0000000..b995b66 --- /dev/null +++ b/docs/probing_samples/vddk_aio_bufsize_probe.py @@ -0,0 +1,224 @@ +#!/usr/bin/env python3 +"""Probe VDDK ``vixDiskLib.nfcAio.Session.BufSizeIn64KB`` vs NFC read extras. + +Not part of the library. Creates a temp lab VM, runs native VDDK over ``nbd`` +under ``strace``, and prints OPEN_SESSION payloads plus IO reply chunk +lengths. BufSizeIn64KB=1 is 64 KiB; 32 is 2 MiB. +""" + +from __future__ import annotations + +import os +import pickle +import re +import struct +import subprocess +import sys +import tempfile + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) +os.environ.pop("LD_PRELOAD", None) + +from tests.integration import vixdisklib # noqa: E402 +from tests.integration.base import ( # noqa: E402 + SECTOR_SIZE, + create_lab_vm, + destroy_lab_vm, + ensure_vddk_library_path, +) + +_REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) +_VDDK = os.path.join(_REPO, ".vddk") +_AIO_MAGIC = 0xA100DA7A +_AIO_HDR = 16 +_NFC_AIO_MSG_OPEN_SESSION = 2 +_NFC_AIO_MSG_IO = 7 +_NFC_AIO_IO_READ = 1 +# 129 sectors: two 64 KiB-class fragments today. 4097: 2 MiB + 512. +_READS = ((129, "129s"), (4097, "2MiB+512")) + + +def _vddk_config(directory: str, buf_size_in_64kb: int, buf_count: int) -> str: + path = os.path.join(directory, "vddk.config") + log = os.path.join(directory, "vddk.log") + with open(path, "w", encoding="utf-8") as config: + config.write(f"tmpDirectory={directory}\n") + config.write(f"log.fileName={log}\n") + config.write("log.fileLevel=verbose\n") + config.write("vixDiskLib.nfc.LogLevel=4\n") + config.write("vixDiskLib.transport.LogLevel=4\n") + config.write(f"vixDiskLib.nfcAio.Session.BufSizeIn64KB={buf_size_in_64kb}\n") + config.write(f"vixDiskLib.nfcAio.Session.BufCount={buf_count}\n") + return path + + +def _worker(lab_pkl: str, work_dir: str, buf_size_in_64kb: int, buf_count: int) -> None: + ensure_vddk_library_path() + with open(lab_pkl, "rb") as pickle_file: + lab = pickle.load(pickle_file) + config_path = _vddk_config(work_dir, buf_size_in_64kb, buf_count) + handle = vixdisklib.VixDiskLibHandle( + vixdisklib_compatibility_version="8.0", config_path=config_path + ) + kwargs = { + "server_name": lab.host, + "port": lab.port, + "thumbprint": lab.thumbprint, + "username": lab.username, + "password": lab.password, + "vmx_spec": lab.vmx_spec, + "transport_modes": "nbd", + "read_only": True, + } + with ( + handle.connect(**kwargs) as conn, + handle.open( + conn, lab.disk_path, flags=vixdisklib.VIXDISKLIB_FLAG_OPEN_READ_ONLY + ) as disk, + ): + print("transport", handle.get_transport_mode(disk), flush=True) + for n_sectors, label in _READS: + buf = vixdisklib.get_buffer(n_sectors * SECTOR_SIZE) + handle.read(disk, 0, n_sectors, buf) + print(f"read {label} ok", flush=True) + handle.exit() + + +def _decode_strace_hex(quoted: str) -> bytes: + parts = re.findall(r"\\x([0-9a-fA-F]{2})", quoted) + return bytes(int(part, 16) for part in parts) + + +def parse_strace(path: str) -> tuple[list[bytes], list[tuple[int, int, int]]]: + """Return client OPEN_SESSION payloads and **server** read-reply chunks. + + VDDK reads the 16-byte AIO header in one syscall and the payload in + the next, so bytes are concatenated per fd before parsing. + """ + syscall_re = re.compile(r'(read|write|recv|send)\((\d+),\s*"(.*?)"') + writes: dict[int, bytearray] = {} + reads: dict[int, bytearray] = {} + with open(path, encoding="utf-8", errors="replace") as strace_file: + for line in strace_file: + match = syscall_re.search(line) + if not match: + continue + op, fd_s, quoted = match.group(1), match.group(2), match.group(3) + buf = _decode_strace_hex(quoted) + if not buf: + continue + fd = int(fd_s) + bucket = writes if op in ("write", "send") else reads + bucket.setdefault(fd, bytearray()).extend(buf) + + def walk(buf: bytes, collect_open: bool, collect_io: bool) -> None: + offset = 0 + while offset + _AIO_HDR <= len(buf): + magic, msg_type, size, _opid = struct.unpack_from("= 16: + open_sessions.append(payload[:16]) + if collect_io and msg_type == _NFC_AIO_MSG_IO and len(payload) >= 40: + opcode = struct.unpack_from(" list[str]: + keys = ( + "Buffer Size", + "BufCount", + "BufSize", + "AIO session", + "Aio Session", + "maximum session", + "Req. buffer", + ) + lines: list[str] = [] + if not os.path.isfile(log_path): + return lines + with open(log_path, encoding="utf-8", errors="replace") as log_file: + for line in log_file: + if any(key in line for key in keys): + lines.append(line.rstrip()) + return lines + + +def _run_traced(lab_pkl: str, buf_size_in_64kb: int, buf_count: int) -> None: + work_dir = tempfile.mkdtemp(prefix=f"vddk-aio-bufsize-{buf_size_in_64kb}-") + strace_path = os.path.join(work_dir, "nfc.strace") + python = sys.executable + cmd = [ + "strace", + "-f", + "-x", + "-s", + "96", + "-e", + "trace=read,write,readv,writev,send,recv,sendto,recvfrom", + "-o", + strace_path, + python, + __file__, + "--worker", + lab_pkl, + work_dir, + str(buf_size_in_64kb), + str(buf_count), + ] + env = os.environ.copy() + env.pop("LD_PRELOAD", None) + lib_path = env.get("LD_LIBRARY_PATH", "") + env["LD_LIBRARY_PATH"] = _VDDK if not lib_path else f"{_VDDK}:{lib_path}" + print(f"\n=== BufSizeIn64KB={buf_size_in_64kb} BufCount={buf_count} ===") + print("work_dir", work_dir) + proc = subprocess.run(cmd, env=env, check=False, text=True, capture_output=True) + sys.stdout.write(proc.stdout) + sys.stderr.write(proc.stderr) + print("worker exit", proc.returncode) + for line in _interesting_log_lines(os.path.join(work_dir, "vddk.log")): + print("LOG", line) + open_sessions, io_reads = parse_strace(strace_path) + for payload in open_sessions: + ints = struct.unpack(" None: + if "--worker" in sys.argv: + _, lab_pkl, work_dir, buf_size, buf_count = sys.argv[1:] + _worker(lab_pkl, work_dir, int(buf_size), int(buf_count)) + return + ensure_vddk_library_path() + lab = create_lab_vm() + lab_pkl = "/tmp/vddk-aio-bufsize-lab.pkl" + try: + with open(lab_pkl, "wb") as pickle_file: + pickle.dump(lab, pickle_file) + print("lab", lab.disk_path, lab.vm_moref) + for buf_size, buf_count in ((1, 1), (32, 1)): + _run_traced(lab_pkl, buf_size, buf_count) + finally: + destroy_lab_vm(lab) + + +if __name__ == "__main__": + main() diff --git a/docs/reverse_engineering_procedure.md b/docs/reverse_engineering_procedure.md index 0a179e6..b988f5c 100644 --- a/docs/reverse_engineering_procedure.md +++ b/docs/reverse_engineering_procedure.md @@ -250,7 +250,9 @@ What that comparison showed: - Request size stays 44; data is extra after the payload. - VDDK sends **one** request even when `length > 65536`. The server replies with several type-7 messages that share `opId`, each with a - chunk length at payload offset 32 (max 65536). + chunk length at payload offset 32 (max = OPEN_SESSION bufSize; + default 65536). `vixDiskLib.nfcAio.Session.BufSizeIn64KB=32` makes + that 2 MiB (`docs/probing_samples/vddk_aio_bufsize_probe.py`). - Treating offset 36 as `NFC_DISK` (`2`) was a 1-sector coincidence; VDDK repeats the byte length there. - Zeros on the wire are real transferred zeros, not a sparse skip. diff --git a/openvixdisklib/nfc_open.py b/openvixdisklib/nfc_open.py index b26358d..1b25bd5 100644 --- a/openvixdisklib/nfc_open.py +++ b/openvixdisklib/nfc_open.py @@ -25,6 +25,7 @@ import socket import ssl import struct +from dataclasses import dataclass from openvixdisklib import fastlz from openvixdisklib.nfc_auth import NfcAuthSession, _ssl_client_context @@ -34,9 +35,12 @@ NFC_AIO_HDR_SIZE = 16 NFC_SECTOR_SIZE = 512 NFC_PROTOCOL_VERSION = 11 -# Max data bytes in one AIO IO request/reply fragment -# (NfcAioInitSession buffer). +# Max data bytes in one AIO IO request/reply fragment. Sent as +# OPEN_SESSION ``bufSize`` (VDDK ``vixDiskLib.nfcAio.Session.BufSizeIn64KB`` +# times 64 KiB). ESXi read extras use this size; 2 MiB (32) works on +# ESXi 8, 16 MiB and 32 MiB do not. NFC_AIO_BUFFER_SIZE = 65536 +NFC_AIO_BUFFER_COUNT = 1 # Classic NFC message types observed on the wire (uint32 at offset 0). NFC_MSG_SESSION_COMPLETE = 4 @@ -78,6 +82,46 @@ NFC_COMPRESSION_FASTLZ = 2 +@dataclass(frozen=True, slots=True) +class ReadFragment: + """One NFC AIO extra in a packed skip-decompression ``buf``. + + Views into ``buf`` (``buf[offset:offset + length]``) are valid + until the next ``read`` into the same buffer. + + ``dest`` is NFC payload offset 28: the byte offset of this fragment + **inside this uncompressed read**, starting at 0. It is not a disk + LBA and not a byte offset from the start of the VMDK. The disk byte + address is ``start_sector * sector_size + dest``. + + ``offset`` is where this extra sits in packed ``buf`` (densely from + 0 in receive order). ``length`` is the extra on the wire. + ``uncompressed_length`` is NFC payload offset 32. + """ + + dest: int + uncompressed_length: int + compression_type: int + offset: int + length: int + + +@dataclass(frozen=True, slots=True) +class ReadResult: + """Outcome of ``NfcDisk.readinto`` / ``VixDiskLibHandle.read``. + + ``compressed_length`` is bytes of extra on the wire. With + ``skip_decompression=True``, extras are packed in ``buf`` from + offset 0 and ``fragments`` describes them. The decompressing path + sets ``fragments`` to empty so callers can still use the lengths + for metrics. + """ + + uncompressed_length: int + compressed_length: int + fragments: tuple[ReadFragment, ...] + + class NfcProtocolError(ConnectionError): """Raised when an NFC message is malformed or reports failure.""" @@ -162,6 +206,20 @@ def _writable_bytes(buf: bytearray | memoryview, length: int) -> memoryview: return raw[:length] +def _aio_extra_len(ctype: int, body: bytes, chunk_len: int) -> int: + """Return this fragment's extra size on the wire.""" + if ctype == NFC_COMPRESSION_FASTLZ: + extra_len = struct.unpack_from(" None: if len(body) > NFC_MSG_SIZE - 4: raise ValueError("NFC classic message body too large") @@ -200,6 +258,8 @@ def __init__( handle: int, sector_size: int, compression: int = NFC_COMPRESSION_NONE, + aio_buffer_size: int = NFC_AIO_BUFFER_SIZE, + aio_buffer_count: int = NFC_AIO_BUFFER_COUNT, ) -> None: """Wrap an AIO session that already has ``path`` open. @@ -210,6 +270,11 @@ def __init__( sector_size: Sector size from the OPEN_FILE reply. compression: NFC IO compression type (``NFC_COMPRESSION_NONE`` or ``NFC_COMPRESSION_FASTLZ``). + aio_buffer_size: OPEN_SESSION extra size in bytes (default + ``NFC_AIO_BUFFER_SIZE``, 64 KiB). ESXi read extras are + at most this large. + aio_buffer_count: OPEN_SESSION buffer pool count (default + ``NFC_AIO_BUFFER_COUNT``). """ self._sock = sock self._op_id = 0 @@ -217,6 +282,8 @@ def __init__( self.handle = handle self.sector_size = sector_size self.compression = compression + self.aio_buffer_size = aio_buffer_size + self.aio_buffer_count = aio_buffer_count self._closed = False def _next_op_id(self) -> int: @@ -264,7 +331,7 @@ def read(self, start_sector: int, num_sectors: int = 1) -> bytes: """Read ``num_sectors`` starting at ``start_sector``. Matches ``VixDiskLib_Read``: one ``NFC_AIO_MSG_IO`` request in - byte units. If the length exceeds the AIO buffer (64 KiB) the + byte units. If the length exceeds the session AIO buffer the server replies with several same-``opId`` fragments, which are placed by the fragment byte offset in the reply (they may arrive out of order). FASTLZ open requests compression in the opcode; @@ -285,22 +352,30 @@ def readinto( start_sector: int, num_sectors: int, buf: bytearray | memoryview, - ) -> int: + skip_decompression: bool = False, + ) -> ReadResult: """Read ``num_sectors`` into ``buf`` starting at ``start_sector``. Uncompressed fragments are received directly into ``buf``. FastLZ still decompresses into a temporary buffer, then copies the - result. ``buf`` must be writable and at least - ``num_sectors * sector_size`` bytes (a ``get_buffer`` ctypes - array is wrapped with ``memoryview`` by the VDDK-shaped handle). + result, unless ``skip_decompression`` is set. ``buf`` must be + writable and at least ``num_sectors * sector_size`` bytes (a + ``get_buffer`` ctypes array is wrapped with ``memoryview`` by + the VDDK-shaped handle). Args: start_sector: Sector offset from the start of the disk. num_sectors: Number of sectors to read. buf: Destination buffer. + skip_decompression: When True, pack NFC extras densely from + offset 0 without FastLZ decompress. Fragment metadata + is in the returned ``ReadResult``. Completion still + uses uncompressed chunk lengths. Returns: - The number of bytes written to ``buf``. + Lengths of the uncompressed request and of extras on the + wire. ``fragments`` is populated only when skipping + decompression. """ if num_sectors < 1: raise ValueError("num_sectors must be at least 1") @@ -314,6 +389,9 @@ def readinto( op_id = self._next_op_id() self._sock.sendall(_pack_aio_hdr(NFC_AIO_MSG_IO, len(payload), op_id) + payload) filled = 0 + wire_bytes = 0 + packed_offset = 0 + fragments: list[ReadFragment] = [] seen: set[int] = set() while filled < length: rhdr = _recvn(self._sock, NFC_AIO_HDR_SIZE) @@ -340,10 +418,26 @@ def readinto( ) seen.add(dest) ctype = opcode >> 32 - chunk_view = data[dest : dest + chunk_len] - if ctype == NFC_COMPRESSION_FASTLZ: - comp_len = struct.unpack_from(" length: + raise NfcProtocolError( + f"packed extras {end} bytes exceed request {length}" + ) + _recvn_into(self._sock, data[packed_offset:end]) + fragments.append( + ReadFragment( + dest=dest, + uncompressed_length=chunk_len, + compression_type=ctype, + offset=packed_offset, + length=extra_len, + ) + ) + packed_offset = end + elif ctype == NFC_COMPRESSION_FASTLZ: + extra = _recvn(self._sock, extra_len) try: chunk = fastlz.decompress(extra, chunk_len) except ValueError as exc: @@ -354,19 +448,24 @@ def readinto( raise NfcProtocolError( f"FastLZ read got {len(chunk)} bytes, expected {chunk_len}" ) - chunk_view[:] = chunk + data[dest : dest + chunk_len] = chunk elif ctype == NFC_COMPRESSION_NONE: - _recvn_into(self._sock, chunk_view) + _recvn_into(self._sock, data[dest : dest + chunk_len]) else: raise NfcProtocolError(f"unsupported NFC IO compression type {ctype}") + wire_bytes += extra_len filled += chunk_len - return length + return ReadResult( + uncompressed_length=length, + compressed_length=wire_bytes, + fragments=tuple(fragments) if skip_decompression else (), + ) def write(self, start_sector: int, num_sectors: int, data: bytes) -> None: """Write ``num_sectors`` starting at ``start_sector``. Matches ``VixDiskLib_Write``: one ``NFC_AIO_MSG_IO`` ``opId`` - for the whole call. Chunks larger than the AIO buffer (64 KiB) + for the whole call. Chunks larger than the session AIO buffer are extra fragments with that same ``opId``; the server replies once. FASTLZ open compresses each fragment when that shrinks it. @@ -384,7 +483,7 @@ def write(self, start_sector: int, num_sectors: int, data: bytes) -> None: op_id = self._next_op_id() frag_offset = 0 while frag_offset < length: - chunk = data[frag_offset : frag_offset + NFC_AIO_BUFFER_SIZE] + chunk = data[frag_offset : frag_offset + self.aio_buffer_size] extra = chunk extra_len = len(chunk) ctype = NFC_COMPRESSION_NONE @@ -484,7 +583,10 @@ def _handshake(sock: socket.socket, client_name: str, op_id: str, version: int) def _aio_prepare(disk: NfcDisk) -> None: - disk._aio_roundtrip(NFC_AIO_MSG_OPEN_SESSION, bytes(16)) + open_session = struct.pack( + " NfcDisk: """Open ``disk_path`` over the authenticated authd socket. @@ -528,7 +632,13 @@ def open_disk( read_only: When True, open with VDDK's read-only NFC flags. compression: ``NFC_COMPRESSION_NONE`` or ``NFC_COMPRESSION_FASTLZ``. OPEN_FILE flags are unchanged; compression is per IO message. + aio_buffer_size: Extra size advertised in OPEN_SESSION (bytes). + aio_buffer_count: Buffer pool count advertised in OPEN_SESSION. """ + if aio_buffer_size < 1: + raise ValueError("aio_buffer_size must be at least 1") + if aio_buffer_count < 1: + raise ValueError("aio_buffer_count must be at least 1") if compression not in (NFC_COMPRESSION_NONE, NFC_COMPRESSION_FASTLZ): raise NotImplementedError( f"NFC compression type {compression} is not supported" @@ -546,6 +656,8 @@ def open_disk( handle=0, sector_size=NFC_SECTOR_SIZE, compression=compression, + aio_buffer_size=aio_buffer_size, + aio_buffer_count=aio_buffer_count, ) _aio_prepare(disk) path_b = disk_path.encode("utf-8") diff --git a/openvixdisklib/openvixdisklib.py b/openvixdisklib/openvixdisklib.py index da24080..cafb7f9 100644 --- a/openvixdisklib/openvixdisklib.py +++ b/openvixdisklib/openvixdisklib.py @@ -26,6 +26,9 @@ from openvixdisklib import nfc_auth, nfc_open +ReadResult = nfc_open.ReadResult +ReadFragment = nfc_open.ReadFragment + LOG = logging.getLogger(__name__) VIXDISKLIB_VERSION_MAJOR = 8 @@ -264,6 +267,8 @@ def open( conn: _Connection, disk_path: str, flags: int = VIXDISKLIB_FLAG_OPEN_READ_ONLY, + aio_buffer_size: int = nfc_open.NFC_AIO_BUFFER_SIZE, + aio_buffer_count: int = nfc_open.NFC_AIO_BUFFER_COUNT, ) -> Iterator[_DiskHandle]: """Open ``disk_path`` over NFC. Matches ``VixDiskLib_Open``. @@ -280,6 +285,13 @@ def open( the disk read-only; omit it for write. ``VIXDISKLIB_FLAG_OPEN_COMPRESSION_FASTLZ`` compresses NFC IO. zlib and skipz are not implemented. + aio_buffer_size: NFC AIO extra size in bytes, advertised in + OPEN_SESSION. Default 64 KiB. ESXi 8 accepts 2 MiB + (``2097152``) and rejects 16 MiB and 32 MiB. This is an + OpenVixDiskLib extension (VDDK uses + ``vixDiskLib.nfcAio.Session.BufSizeIn64KB``). + aio_buffer_count: NFC AIO buffer pool count. Default 1. + VDDK's default is 4. """ LOG.debug("Openning VixDiskLib disk: %s", disk_path) compression = _nfc_compression(flags) @@ -298,7 +310,12 @@ def open( session = nfc_auth.NfcAuthSession(conn.si, ticket, authd_sock, nfc_ssl=nfc_ssl) try: disk = nfc_open.open_disk( - session, disk_path, read_only=read_only, compression=compression + session, + disk_path, + read_only=read_only, + compression=compression, + aio_buffer_size=aio_buffer_size, + aio_buffer_count=aio_buffer_count, ) except Exception: authd_sock.close() @@ -315,7 +332,8 @@ def read( start_sector: int, num_sectors: int, buf: ctypes.Array | bytearray | memoryview, - ) -> None: + skip_decompression: bool = False, + ) -> ReadResult: """Read ``num_sectors`` from ``start_sector`` into ``buf``. Args: @@ -323,9 +341,27 @@ def read( start_sector: First sector to read. num_sectors: Number of sectors to read. buf: Destination buffer (``get_buffer`` or a writable bytes-like). - Uncompressed NFC extra is received into this buffer. + Uncompressed NFC extra is received into this buffer + unless ``skip_decompression`` is True. + skip_decompression: OpenVixDiskLib extension. When True, + pack NFC extras densely from offset 0 without FastLZ + decompress. ``ReadResult.fragments`` lists each extra + (``ReadFragment``). ``ReadFragment.dest`` is the byte + offset inside this uncompressed read, not a disk + offset. Completion still uses uncompressed + lengths. With no FASTLZ open flag this only records + raw extras (``compressed_length == uncompressed_length``). + + Returns: + Uncompressed and wire lengths. ``fragments`` is empty unless + ``skip_decompression`` is True. """ - disk_handle.disk.readinto(start_sector, num_sectors, memoryview(buf)) + return disk_handle.disk.readinto( + start_sector, + num_sectors, + memoryview(buf), + skip_decompression=skip_decompression, + ) def write( self, diff --git a/tests/integration/test_nfc_read_write.py b/tests/integration/test_nfc_read_write.py index ff27728..b435eb3 100644 --- a/tests/integration/test_nfc_read_write.py +++ b/tests/integration/test_nfc_read_write.py @@ -10,6 +10,9 @@ from openvixdisklib import nfc_open from tests.integration.base import SECTOR_SIZE, LabEnv, pattern_bytes +_1MIB = 1024 * 1024 +_2MIB = 2 * 1024 * 1024 +_16MIB = 16 * 1024 * 1024 _32MIB = 32 * 1024 * 1024 @@ -70,14 +73,60 @@ def test_sector_writes_and_reads(self, lab: LabEnv, compression: int) -> None: [nfc_open.NFC_COMPRESSION_NONE, nfc_open.NFC_COMPRESSION_FASTLZ], ids=["plain", "fastlz"], ) - def test_write_and_read_32mb(self, lab: LabEnv, compression: int) -> None: - """Write 32 MiB (512 AIO chunks) and read it back in one request.""" + @pytest.mark.parametrize( + "aio_buffer_count, aio_buffer_size", + [ + (1, nfc_open.NFC_AIO_BUFFER_SIZE), + (1, _1MIB), + (1, _2MIB), + (4, _2MIB), + pytest.param( + 1, + _16MIB, + marks=pytest.mark.xfail( + raises=nfc_open.NfcProtocolError, + reason="ESXi 8 rejects OPEN_SESSION bufSize 16 MiB", + strict=True, + ), + ), + pytest.param( + 1, + _32MIB, + marks=pytest.mark.xfail( + raises=nfc_open.NfcProtocolError, + reason="ESXi 8 rejects OPEN_SESSION bufSize 32 MiB", + strict=True, + ), + ), + ], + ids=[ + "count1-64kib", + "count1-1mib", + "count1-2mib", + "count4-2mib", + "count1-16mib", + "count1-32mib", + ], + ) + def test_write_and_read_32mb( + self, + lab: LabEnv, + compression: int, + aio_buffer_count: int, + aio_buffer_size: int, + ) -> None: + """Write 32 MiB and read it back for several OPEN_SESSION sizes.""" n_sectors = _32MIB // SECTOR_SIZE to_write = os.urandom(_32MIB) with ( lab.authenticate(read_only=False) as session, nfc_open.open_disk( - session, lab.disk_path, read_only=False, compression=compression + session, + lab.disk_path, + read_only=False, + compression=compression, + aio_buffer_size=aio_buffer_size, + aio_buffer_count=aio_buffer_count, ) as disk, ): disk.write(0, n_sectors, to_write) diff --git a/tests/integration/test_openvixdisklib.py b/tests/integration/test_openvixdisklib.py index 39b130b..6c853d6 100644 --- a/tests/integration/test_openvixdisklib.py +++ b/tests/integration/test_openvixdisklib.py @@ -3,11 +3,15 @@ """Exercise the VDDK-compatible openvixdisklib handle against the lab.""" +from typing import Any + import pytest from pyVim.connect import Disconnect from pyVmomi import vim +from openvixdisklib import fastlz, nfc_open from openvixdisklib import openvixdisklib as vixdisklib +from openvixdisklib.openvixdisklib import ReadResult from tests.integration.base import ( SECTOR_AT_1GB, SECTOR_SIZE, @@ -28,6 +32,29 @@ def _virtual_disk_backing( raise AssertionError(f"{vm._moId} has no virtual disk") +_2MIB = 2 * 1024 * 1024 + + +def _rebuild_skip(buf: Any, result: ReadResult) -> bytes: + """Decompress packed skip-decompression extras into uncompressed bytes.""" + view = buf.raw if hasattr(buf, "raw") else buf + out = bytearray(result.uncompressed_length) + packed = 0 + for frag in result.fragments: + extra = bytes(view[frag.offset : frag.offset + frag.length]) + packed += frag.length + if frag.compression_type == nfc_open.NFC_COMPRESSION_FASTLZ: + chunk = fastlz.decompress(extra, frag.uncompressed_length) + elif frag.compression_type == nfc_open.NFC_COMPRESSION_NONE: + chunk = extra + else: + raise AssertionError(f"unexpected compression_type {frag.compression_type}") + assert len(chunk) == frag.uncompressed_length + out[frag.dest : frag.dest + frag.uncompressed_length] = chunk + assert packed == result.compressed_length + return bytes(out) + + class TestOpenvixdisklib: @pytest.mark.parametrize("transport_mode", ["nbdssl", "nbd"]) @pytest.mark.parametrize( @@ -134,3 +161,62 @@ def read_sector(path: str) -> bytes: _wait_for_task(vm.RemoveAllSnapshots_Task()) finally: Disconnect(si) + + @pytest.mark.parametrize( + "aio_buffer_size, n_sectors, n_fragments", + [ + (nfc_open.NFC_AIO_BUFFER_SIZE, 128, 1), + (nfc_open.NFC_AIO_BUFFER_SIZE, 129, 2), + (_2MIB, 129, 1), + ], + ids=["64kib-128s", "64kib-129s", "2mib-129s"], + ) + def test_skip_decompression_fastlz( + self, + lab: LabEnv, + aio_buffer_size: int, + n_sectors: int, + n_fragments: int, + ) -> None: + """Pack FastLZ extras and rebuild the same bytes as a normal read.""" + length = n_sectors * SECTOR_SIZE + expected = pattern_bytes(length, b"OVDL-SKIP-") + handle = vixdisklib.VixDiskLibHandle( + vixdisklib_compatibility_version="8.0", config_path=None + ) + write_buf = vixdisklib.get_buffer(length) + plain_buf = vixdisklib.get_buffer(length) + skip_buf = vixdisklib.get_buffer(length) + write_buf[:length] = expected + connect_kwargs = lab.vixdisklib_connect_kwargs( + {"allow_untrusted": lab.allow_untrusted, "transport_modes": "nbd"} + ) + flags = vixdisklib.VIXDISKLIB_FLAG_OPEN_COMPRESSION_FASTLZ + with ( + handle.connect(**connect_kwargs) as conn, + handle.open( + conn, + lab.disk_path, + flags=flags, + aio_buffer_size=aio_buffer_size, + aio_buffer_count=1, + ) as disk, + ): + handle.write(disk, 0, n_sectors, write_buf) + plain = handle.read(disk, 0, n_sectors, plain_buf) + skip = handle.read(disk, 0, n_sectors, skip_buf, skip_decompression=True) + assert isinstance(plain, ReadResult) + assert plain.fragments == () + assert plain.uncompressed_length == length + assert plain.compressed_length <= length + assert skip.uncompressed_length == length + assert skip.compressed_length <= length + assert skip.compressed_length == plain.compressed_length + assert len(skip.fragments) == n_fragments + dests = {frag.dest for frag in skip.fragments} + if n_fragments == 1: + assert dests == {0} + else: + assert dests == {0, nfc_open.NFC_AIO_BUFFER_SIZE} + assert plain_buf.raw[:length] == expected + assert _rebuild_skip(skip_buf, skip) == expected diff --git a/tests/perf/test_compare.py b/tests/perf/test_compare.py index cb75f91..2b608d3 100644 --- a/tests/perf/test_compare.py +++ b/tests/perf/test_compare.py @@ -5,18 +5,71 @@ from __future__ import annotations +import json +import os +import pickle +import subprocess +import sys +import tempfile import time from typing import Any +from openvixdisklib import nfc_open from openvixdisklib import openvixdisklib as open_vix from tests.integration import vixdisklib -from tests.integration.base import SECTOR_SIZE, LabEnv, pattern_bytes +from tests.integration.base import ( + SECTOR_SIZE, + LabEnv, + ensure_vddk_library_path, + pattern_bytes, +) +_REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) _SIZES = ( ("64KiB", 64 * 1024), ("129 sectors", 129 * SECTOR_SIZE), ("32MiB", 32 * 1024 * 1024), ) +_1MIB = 1024 * 1024 +_2MIB = 2 * 1024 * 1024 +# ESXi 8 accepts 64 KiB, 1 MiB, and 2 MiB extras. 16 MiB and 32 MiB +# OPEN_SESSION are rejected (AIO error). Broadcom's 16 MiB cap is +# size×count session memory, not a larger extra; VDDK's per-buffer max +# is 2 MiB (``BufSizeIn64KB=16`` is 1 MiB). +_AIO_SESSIONS = ( + (nfc_open.NFC_AIO_BUFFER_COUNT, nfc_open.NFC_AIO_BUFFER_SIZE), + (1, _1MIB), + (1, _2MIB), + (4, _2MIB), +) +_64KIB = 64 * 1024 + + +def _aio_size_label(nbytes: int) -> str: + """Return a short label for an AIO extra size.""" + if nbytes % (1024 * 1024) == 0: + return f"{nbytes // (1024 * 1024)}MiB" + if nbytes % 1024 == 0: + return f"{nbytes // 1024}KiB" + return str(nbytes) + + +def _vddk_aio_config( + directory: str, aio_buffer_size: int, aio_buffer_count: int +) -> str: + """Write a temp VDDK config for ``BufSizeIn64KB`` and ``BufCount``.""" + if aio_buffer_size % _64KIB: + raise ValueError( + f"VDDK BufSizeIn64KB needs a 64 KiB multiple, got {aio_buffer_size}" + ) + path = os.path.join(directory, "vddk.config") + with open(path, "w", encoding="utf-8") as config: + config.write(f"tmpDirectory={directory}\n") + config.write( + f"vixDiskLib.nfcAio.Session.BufSizeIn64KB={aio_buffer_size // _64KIB}\n" + ) + config.write(f"vixDiskLib.nfcAio.Session.BufCount={aio_buffer_count}\n") + return path def _connect_extra(lab: LabEnv, module: Any, transport_mode: str) -> dict[str, Any]: @@ -27,37 +80,182 @@ def _connect_extra(lab: LabEnv, module: Any, transport_mode: str) -> dict[str, A return extra -def _time_write_read( +def _time_write_read_once( lab: LabEnv, module: Any, payload: bytes, flags: int = 0, transport_mode: str = "nbdssl", + aio_buffer_size: int = nfc_open.NFC_AIO_BUFFER_SIZE, + aio_buffer_count: int = nfc_open.NFC_AIO_BUFFER_COUNT, + config_dir: str | None = None, + skip_decompression: bool = False, ) -> tuple[float, float]: - """Write ``payload`` at sector 0, read it back, and return durations.""" + """Write ``payload`` at sector 0, read it back, and return durations. + + Does not call ``VixDiskLib_Exit``. Native VDDK double-frees if + ``InitEx``/``Exit`` are paired more than once in the same process. + ``skip_decompression`` is OpenVixDiskLib FastLZ skip; ``buf`` then + holds packed extras, not sector bytes. + """ n_sectors = len(payload) // SECTOR_SIZE + config_path = None + if module is vixdisklib: + if config_dir is None: + raise ValueError("VDDK timings need a config_dir") + config_path = _vddk_aio_config(config_dir, aio_buffer_size, aio_buffer_count) handle = module.VixDiskLibHandle( - vixdisklib_compatibility_version="8.0", config_path=None + vixdisklib_compatibility_version="8.0", config_path=config_path ) write_buf = module.get_buffer(len(payload)) read_buf = module.get_buffer(len(payload)) write_buf[: len(payload)] = payload kwargs = lab.vixdisklib_connect_kwargs(_connect_extra(lab, module, transport_mode)) + open_kwargs: dict[str, Any] = {"flags": flags} + if module is open_vix: + open_kwargs["aio_buffer_size"] = aio_buffer_size + open_kwargs["aio_buffer_count"] = aio_buffer_count with ( handle.connect(**kwargs) as conn, - handle.open(conn, lab.disk_path, flags=flags) as disk, + handle.open(conn, lab.disk_path, **open_kwargs) as disk, ): started = time.perf_counter() handle.write(disk, 0, n_sectors, write_buf) write_s = time.perf_counter() - started read_buf[: len(payload)] = b"\xa5" * len(payload) + read_kwargs: dict[str, Any] = {} + if skip_decompression: + read_kwargs["skip_decompression"] = True started = time.perf_counter() - handle.read(disk, 0, n_sectors, read_buf) + result = handle.read(disk, 0, n_sectors, read_buf, **read_kwargs) read_s = time.perf_counter() - started - assert read_buf.raw[: len(payload)] == payload + if skip_decompression: + assert result.uncompressed_length == len(payload) + assert result.compressed_length <= len(payload) + assert result.fragments + else: + assert read_buf.raw[: len(payload)] == payload return write_s, read_s +def _run_vddk_worker(lab_pkl: str, job_pkl: str, work_dir: str) -> None: + """InitEx once in this process, time one write/read, write result.json.""" + os.environ.pop("LD_PRELOAD", None) + ensure_vddk_library_path() + with open(lab_pkl, "rb") as pickle_file: + lab = pickle.load(pickle_file) + with open(job_pkl, "rb") as pickle_file: + job = pickle.load(pickle_file) + payload = pattern_bytes(job["nbytes"], f"PERF-{job['label']}-".encode()) + write_s, read_s = _time_write_read_once( + lab, + vixdisklib, + payload, + flags=job["flags"], + transport_mode=job["transport_mode"], + aio_buffer_size=job["aio_buffer_size"], + aio_buffer_count=job["aio_buffer_count"], + config_dir=work_dir, + ) + result_path = os.path.join(work_dir, "result.json") + with open(result_path, "w", encoding="utf-8") as result_file: + json.dump({"write_s": write_s, "read_s": read_s}, result_file) + + +def _time_vddk_subprocess( + lab: LabEnv, + label: str, + nbytes: int, + flags: int, + transport_mode: str, + aio_buffer_size: int, + aio_buffer_count: int, +) -> tuple[float, float]: + """Time native VDDK in a child process so InitEx sees this AIO config.""" + with tempfile.TemporaryDirectory(prefix="vddk-perf-") as work_dir: + lab_pkl = os.path.join(work_dir, "lab.pkl") + job_pkl = os.path.join(work_dir, "job.pkl") + result_path = os.path.join(work_dir, "result.json") + with open(lab_pkl, "wb") as pickle_file: + pickle.dump(lab, pickle_file) + with open(job_pkl, "wb") as pickle_file: + pickle.dump( + { + "label": label, + "nbytes": nbytes, + "flags": flags, + "transport_mode": transport_mode, + "aio_buffer_size": aio_buffer_size, + "aio_buffer_count": aio_buffer_count, + }, + pickle_file, + ) + env = os.environ.copy() + env.pop("LD_PRELOAD", None) + pythonpath = env.get("PYTHONPATH", "") + env["PYTHONPATH"] = _REPO if not pythonpath else f"{_REPO}:{pythonpath}" + proc = subprocess.run( + [ + sys.executable, + os.path.abspath(__file__), + "--vddk-worker", + lab_pkl, + job_pkl, + work_dir, + ], + check=False, + capture_output=True, + text=True, + env=env, + cwd=_REPO, + ) + if proc.returncode != 0 or not os.path.exists(result_path): + raise RuntimeError( + "VDDK perf worker failed " + f"(exit {proc.returncode}): {proc.stderr}\n{proc.stdout}" + ) + with open(result_path, encoding="utf-8") as result_file: + result = json.load(result_file) + return float(result["write_s"]), float(result["read_s"]) + + +def _time_write_read( + lab: LabEnv, + module: Any, + label: str, + nbytes: int, + flags: int = 0, + transport_mode: str = "nbdssl", + aio_buffer_size: int = nfc_open.NFC_AIO_BUFFER_SIZE, + aio_buffer_count: int = nfc_open.NFC_AIO_BUFFER_COUNT, + skip_decompression: bool = False, +) -> tuple[float, float]: + """Time one write/read; native VDDK runs in a subprocess.""" + if skip_decompression and module is vixdisklib: + raise ValueError("skip_decompression is OpenVixDiskLib-only") + if module is vixdisklib: + return _time_vddk_subprocess( + lab, + label, + nbytes, + flags, + transport_mode, + aio_buffer_size, + aio_buffer_count, + ) + payload = pattern_bytes(nbytes, f"PERF-{label}-".encode()) + return _time_write_read_once( + lab, + module, + payload, + flags=flags, + transport_mode=transport_mode, + aio_buffer_size=aio_buffer_size, + aio_buffer_count=aio_buffer_count, + skip_decompression=skip_decompression, + ) + + def _mib_per_s(nbytes: int, seconds: float) -> float: if seconds <= 0: return float("inf") @@ -66,49 +264,74 @@ def _mib_per_s(nbytes: int, seconds: float) -> float: class TestCompare: def test_write_read_throughput(self, lab: LabEnv, vddk: None) -> None: - """Time matching write/read sizes on VDDK and openvixdisklib.""" + """Time matching write/read sizes on VDDK and openvixdisklib. + + Prints ``aio_size`` / ``aio_count`` for each OPEN_SESSION + (64 KiB×1, 1 MiB×1, 2 MiB×1, 2 MiB×4). VDDK gets those via + ``vixDiskLib.nfcAio.Session.BufSizeIn64KB`` / ``BufCount`` in a + fresh process per row (``VixDiskLib_Exit`` is not loop-safe). + ``fastlz-skip`` is OpenVixDiskLib ``skip_decompression`` (packed + extras, no FastLZ decode); VDDK has no equivalent. + """ libraries = ( ("vddk", vixdisklib), ("openvixdisklib", open_vix), ) transports = ("nbdssl", "nbd") open_modes = ( - ("plain", 0), - ("fastlz", vixdisklib.VIXDISKLIB_FLAG_OPEN_COMPRESSION_FASTLZ), + ("plain", 0, False), + ("fastlz", vixdisklib.VIXDISKLIB_FLAG_OPEN_COMPRESSION_FASTLZ, False), + ( + "fastlz-skip", + vixdisklib.VIXDISKLIB_FLAG_OPEN_COMPRESSION_FASTLZ, + True, + ), ) - rows: list[tuple[str, str, str, str, float, float, float, float]] = [] + rows: list[tuple[str, str, int, str, str, str, float, float, float, float]] = [] for label, nbytes in _SIZES: - payload = pattern_bytes(nbytes, f"PERF-{label}-".encode()) - for transport_mode in transports: - for mode_name, flags in open_modes: - for name, module in libraries: - write_s, read_s = _time_write_read( - lab, - module, - payload, - flags=flags, - transport_mode=transport_mode, - ) - rows.append( - ( + for aio_buffer_count, aio_buffer_size in _AIO_SESSIONS: + aio_label = _aio_size_label(aio_buffer_size) + for transport_mode in transports: + for mode_name, flags, skip_decompression in open_modes: + for name, module in libraries: + if skip_decompression and module is vixdisklib: + continue + write_s, read_s = _time_write_read( + lab, + module, label, - transport_mode, - mode_name, - name, - write_s, - read_s, - _mib_per_s(nbytes, write_s), - _mib_per_s(nbytes, read_s), + nbytes, + flags=flags, + transport_mode=transport_mode, + aio_buffer_size=aio_buffer_size, + aio_buffer_count=aio_buffer_count, + skip_decompression=skip_decompression, + ) + rows.append( + ( + label, + aio_label, + aio_buffer_count, + transport_mode, + mode_name, + name, + write_s, + read_s, + _mib_per_s(nbytes, write_s), + _mib_per_s(nbytes, read_s), + ) ) - ) print() print( - f"{'size':<14} {'transport':<10} {'flags':<8} {'library':<16} " + f"{'size':<14} {'aio_size':<8} {'aio_count':>9} " + f"{'transport':<10} {'flags':<12} {'library':<16} " f"{'write_s':>10} {'read_s':>10} " f"{'write_MiB/s':>12} {'read_MiB/s':>12}" ) for ( label, + aio_label, + aio_buffer_count, transport_mode, mode_name, name, @@ -118,7 +341,16 @@ def test_write_read_throughput(self, lab: LabEnv, vddk: None) -> None: read_r, ) in rows: print( - f"{label:<14} {transport_mode:<10} {mode_name:<8} {name:<16} " + f"{label:<14} {aio_label:<8} {aio_buffer_count:>9} " + f"{transport_mode:<10} {mode_name:<12} {name:<16} " f"{write_s:10.3f} {read_s:10.3f} " f"{write_r:12.1f} {read_r:12.1f}" ) + + +if __name__ == "__main__": + if sys.argv[1:2] == ["--vddk-worker"]: + _, _, lab_pkl, job_pkl, work_dir = sys.argv + _run_vddk_worker(lab_pkl, job_pkl, work_dir) + else: + raise SystemExit("usage: test_compare.py --vddk-worker LAB JOB DIR")