From 20995912459cb1a874dcdf778108eca24fd7f179 Mon Sep 17 00:00:00 2001 From: Lucian Petrut Date: Thu, 10 Sep 2026 12:26:52 +0000 Subject: [PATCH] Allow retrieving compressed chunks VDDK always decompresses the chunks it retrieves. However, some callers may be interested in the compressed chunks, which may be forwarded to another service involved in the backup process. We'll add a read flag (skip_decompression). If set, the read operation will return a "ReadResult" container, containing a list of fragments and their compressed / decompressed lengths. If the returned buffer size matches the AIO buffer size (which now becomes configurable), at most one fragment will be returned. While at it, we're adding perf tests that check various AIO buffer sizes, cross checking against VDDK. --- AGENTS.md | 2 +- README.md | 9 +- docs/nfc_open.md | 39 ++- docs/nfc_read.md | 86 ++++- docs/nfc_write.md | 11 +- .../probing_samples/vddk_aio_bufsize_probe.py | 224 +++++++++++++ docs/reverse_engineering_procedure.md | 4 +- openvixdisklib/nfc_open.py | 148 +++++++-- openvixdisklib/openvixdisklib.py | 44 ++- tests/integration/test_nfc_read_write.py | 55 +++- tests/integration/test_openvixdisklib.py | 86 +++++ tests/perf/test_compare.py | 300 ++++++++++++++++-- 12 files changed, 914 insertions(+), 94 deletions(-) create mode 100644 docs/probing_samples/vddk_aio_bufsize_probe.py 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")