[KV Offload][FS] Opt-in per-block CRC32C checksums

csrc/fs_io.cpp

@@ -5,22 +5,137 @@
 
 #include <errno.h>
 #include <fcntl.h>
+#include <sys/xattr.h>
 #include <unistd.h>
 
+#include <array>
+#include <cstdint>
+#include <cstring>
 #include <filesystem>
 #include <string>
 #include <vector>
 
+#if defined(__x86_64__)
+  #include <nmmintrin.h>
+#endif
+
 #if defined(O_DIRECT)
 constexpr int kODirectFlag = O_DIRECT;
 #else
 constexpr int kODirectFlag = 0;
 #endif
 
+// Extended attribute holding a block's CRC32C as four big-endian bytes. Must
+// match _CHECKSUM_XATTR in vllm/v1/kv_offload/tiering/fs/io.py.
+constexpr const char* kChecksumXattr = "user.vllm.kv_crc32c";
+constexpr size_t kChecksumSize = 4;
+
+// errno of fgetxattr when the attribute is absent.
+#if defined(__APPLE__)
+constexpr int kXattrAbsentErrno = ENOATTR;
+#else
+constexpr int kXattrAbsentErrno = ENODATA;
+#endif
+
+// Returned by verify_checksum for a mismatch; distinct from any errno.
+constexpr int kChecksumMismatch = -1;
+
 extern "C" {
 
 namespace {
 
+// CRC32C (Castagnoli) polynomial, bit-reflected.
+constexpr uint32_t kCrc32cPolyReflected = 0x82F63B78;
+
+constexpr std::array<uint32_t, 256> kCrc32cTable = [] {
+  std::array<uint32_t, 256> table{};
+  for (uint32_t i = 0; i < 256; i++) {
+    uint32_t crc = i;
+    for (int bit = 0; bit < 8; bit++) {
+      crc = (crc & 1) ? (crc >> 1) ^ kCrc32cPolyReflected : crc >> 1;
+    }
+    table[i] = crc;
+  }
+  return table;
+}();
+
+inline uint32_t crc32c_update_sw(uint32_t crc, const uint8_t* data,
+                                 size_t size) {
+  for (size_t i = 0; i < size; i++) {
+    crc = kCrc32cTable[(crc ^ data[i]) & 0xFF] ^ (crc >> 8);
+  }
+  return crc;
+}
+
+#if defined(__x86_64__)
+__attribute__((target("sse4.2"))) inline uint32_t crc32c_update_hw(
+    uint32_t crc, const uint8_t* data, size_t size) {
+  uint64_t crc64 = crc;
+  for (; size >= 8; data += 8, size -= 8) {
+    uint64_t word;
+    std::memcpy(&word, data, sizeof(word));
+    crc64 = _mm_crc32_u64(crc64, word);
+  }
+  crc = static_cast<uint32_t>(crc64);
+  for (; size > 0; data++, size--) {
+    crc = _mm_crc32_u8(crc, *data);
+  }
+  return crc;
+}
+
+inline bool has_crc32c_hw() { return __builtin_cpu_supports("sse4.2"); }
+#endif
+
+// Must match crc32c in vllm/v1/kv_offload/tiering/fs/io.py.
+inline uint32_t crc32c(const char* data, size_t size) {
+  const auto* bytes = reinterpret_cast<const uint8_t*>(data);
+#if defined(__x86_64__)
+  static const bool use_hw = has_crc32c_hw();
+  if (use_hw) {
+    return ~crc32c_update_hw(0xFFFFFFFF, bytes, size);
+  }
+#endif
+  return ~crc32c_update_sw(0xFFFFFFFF, bytes, size);
+}
+
+inline int set_checksum_xattr(int fd, uint32_t crc) {
+  const unsigned char value[kChecksumSize] = {
+      static_cast<unsigned char>(crc >> 24),
+      static_cast<unsigned char>(crc >> 16),
+      static_cast<unsigned char>(crc >> 8), static_cast<unsigned char>(crc)};
+#if defined(__APPLE__)
+  return fsetxattr(fd, kChecksumXattr, value, kChecksumSize, 0, 0);
+#else
+  return fsetxattr(fd, kChecksumXattr, value, kChecksumSize, 0);
+#endif
+}
+
+// Returns 0 if `data` matches the CRC32C recorded on `fd` or none is recorded,
+// kChecksumMismatch on a mismatch, or the errno of a failed lookup.
+inline int verify_checksum(int fd, const char* data, size_t size) {
+  unsigned char value[kChecksumSize];
+#if defined(__APPLE__)
+  const ssize_t len = fgetxattr(fd, kChecksumXattr, value, kChecksumSize, 0, 0);
+#else
+  const ssize_t len = fgetxattr(fd, kChecksumXattr, value, kChecksumSize);
+#endif
+  if (len < 0) {
+    if (errno == kXattrAbsentErrno) {
+      return 0;
+    }
+    // ERANGE: the record is larger than a checksum.
+    return errno == ERANGE ? kChecksumMismatch : errno;
+  }
+  if (len != static_cast<ssize_t>(kChecksumSize)) {
+    return kChecksumMismatch;
+  }
+  const uint32_t expected = (static_cast<uint32_t>(value[0]) << 24) |
+                            (static_cast<uint32_t>(value[1]) << 16) |
+                            (static_cast<uint32_t>(value[2]) << 8) |
+                            static_cast<uint32_t>(value[3]);
+  return crc32c(data, size) == expected ? 0 : kChecksumMismatch;
+}
+
 // Returns 0 on success, or the std::error_code's POSIX-compatible value on
 // failure, mirroring the errno convention used by the syscalls below.
 inline int ensure_parent_dirs(const std::string& path) {
@@ -38,7 +153,8 @@ inline int ensure_parent_dirs(const std::string& path) {
 // before any subsequent cleanup call can overwrite it. On failure, the temp
 // file is removed.
 inline int _store_block(const char* tmp_path, const char* dest_path,
-                        const char* src, size_t size, bool use_o_direct) {
+                        const char* src, size_t size, bool use_o_direct,
+                        bool checksum) {
   if (access(dest_path, F_OK) == 0) {
     return 0;  // Already present.
   }
@@ -62,6 +178,14 @@ inline int _store_block(const char* tmp_path, const char* dest_path,
     return err;
   }
 
+  // Before the rename, so a published block never lacks it.
+  if (checksum && set_checksum_xattr(fd, crc32c(src, size)) != 0) {
+    const int err = errno;
+    close(fd);
+    unlink(tmp_path);
+    return err;
+  }
+
   if (close(fd) != 0) {
     const int err = errno;
     unlink(tmp_path);
@@ -88,8 +212,10 @@ inline int _store_block(const char* tmp_path, const char* dest_path,
 // corruption. Open failures and read errors (bytes_read < 0) are
 // transient/ambiguous and leave the file untouched; a close failure after a
 // full read is harmless and does not fail the load.
+// With `checksum`, a CRC32C mismatch also removes the file and returns
+// kChecksumMismatch.
 inline int _load_block(const char* source_path, char* dst, size_t size,
-                       bool use_o_direct) {
+                       bool use_o_direct, bool checksum) {
   const int o_direct_flag = use_o_direct ? kODirectFlag : 0;
   const int fd = open(source_path, O_RDONLY | o_direct_flag, 0);
   if (fd < 0) {
@@ -110,6 +236,17 @@ inline int _load_block(const char* source_path, char* dst, size_t size,
     return EIO;
   }
 
+  if (checksum) {
+    const int err = verify_checksum(fd, dst, size);
+    if (err != 0) {
+      close(fd);
+      if (err == kChecksumMismatch) {
+        unlink(source_path);
+      }
+      return err;
+    }
+  }
+
   // A close error after a successful full read is harmless: the data is
   // already in the destination buffer, so the load succeeds.
   close(fd);
@@ -203,16 +340,19 @@ static PyObject* batch_lookup(PyObject* /*self*/, PyObject* args) {
 /// @param use_o_direct bool – whether to open files with O_DIRECT
 ///                     (default True). Ignored where O_DIRECT is unsupported
 ///                     by the platform.
+/// @param checksum     bool – attach each block's CRC32C as an extended
+///                     attribute before publishing it (default False).
 /// @note Releases the GIL for the entire batch. Raises on first error.
 static PyObject* batch_store_block(PyObject* /*self*/, PyObject* args) {
   PyObject* tmp_paths_obj = nullptr;
   PyObject* dest_paths_obj = nullptr;
   PyObject* buffers_obj = nullptr;
   int use_o_direct = 1;
+  int checksum = 0;
 
-  if (!PyArg_ParseTuple(args, "O!O!O!|p", &PyList_Type, &tmp_paths_obj,
+  if (!PyArg_ParseTuple(args, "O!O!O!|pp", &PyList_Type, &tmp_paths_obj,
                         &PyList_Type, &dest_paths_obj, &PyList_Type,
-                        &buffers_obj, &use_o_direct)) {
+                        &buffers_obj, &use_o_direct, &checksum)) {
     return nullptr;
   }
 
@@ -241,9 +381,9 @@ static PyObject* batch_store_block(PyObject* /*self*/, PyObject* args) {
   {
     Py_BEGIN_ALLOW_THREADS for (Py_ssize_t i = 0; i < n; i++) {
       const char* buf = static_cast<const char*>(buffers[i].buf);
-      const int err =
-          _store_block(tmp_paths[i], dest_paths[i], buf,
-                       static_cast<size_t>(buffers[i].len), use_o_direct);
+      const int err = _store_block(tmp_paths[i], dest_paths[i], buf,
+                                   static_cast<size_t>(buffers[i].len),
+                                   use_o_direct, checksum);
       if (err != 0) {
         failed_index = i;
         failure_errno = err;
@@ -272,14 +412,18 @@ static PyObject* batch_store_block(PyObject* /*self*/, PyObject* args) {
 /// @param use_o_direct bool – whether to open files with O_DIRECT
 ///                     (default True). Ignored where O_DIRECT is unsupported
 ///                     by the platform.
+/// @param checksum     bool – check each block against its recorded CRC32C
+///                     (default False). A mismatch raises EBADMSG with
+///                     `checksum_failed` set.
 /// @note Releases the GIL for the entire batch. Raises on first error.
 static PyObject* batch_load_block(PyObject* /*self*/, PyObject* args) {
   PyObject* source_paths_obj = nullptr;
   PyObject* buffers_obj = nullptr;
   int use_o_direct = 1;
+  int checksum = 0;
 
-  if (!PyArg_ParseTuple(args, "O!O!|p", &PyList_Type, &source_paths_obj,
-                        &PyList_Type, &buffers_obj, &use_o_direct)) {
+  if (!PyArg_ParseTuple(args, "O!O!|pp", &PyList_Type, &source_paths_obj,
+                        &PyList_Type, &buffers_obj, &use_o_direct, &checksum)) {
     return nullptr;
   }
 
@@ -306,7 +450,7 @@ static PyObject* batch_load_block(PyObject* /*self*/, PyObject* args) {
       char* buf = static_cast<char*>(buffers[i].buf);
       const int err =
           _load_block(source_paths[i], buf, static_cast<size_t>(buffers[i].len),
-                      use_o_direct);
+                      use_o_direct, checksum);
       if (err != 0) {
         failed_index = i;
         failure_errno = err;
@@ -319,8 +463,9 @@ static PyObject* batch_load_block(PyObject* /*self*/, PyObject* args) {
   release_buffer_list(buffers);
 
   if (failed_index >= 0) {
+    const bool checksum_failed = failure_errno == kChecksumMismatch;
     // PyErr_SetFromErrnoWithFilename() reads the errno to format exception.
-    errno = failure_errno;
+    errno = checksum_failed ? EBADMSG : failure_errno;
     PyErr_SetFromErrnoWithFilename(PyExc_OSError, source_paths[failed_index]);
     // Attach the number of blocks that loaded before the failure so the tier
     // can keep them (partial success). failed_index == count of blocks read OK.
@@ -333,6 +478,9 @@ static PyObject* batch_load_block(PyObject* /*self*/, PyObject* args) {
         PyObject_SetAttrString(evalue, "num_succeeded", num);
         Py_DECREF(num);
       }
+      if (checksum_failed) {
+        PyObject_SetAttrString(evalue, "checksum_failed", Py_True);
+      }
     }
     PyErr_Restore(etype, evalue, etb);
     return nullptr;
@@ -349,14 +497,16 @@ static PyMethodDef fs_io_C_methods[] = {
     {"batch_store_block", batch_store_block, METH_VARARGS,
      "batch_store_block(tmp_paths: list[str], dest_paths: list[str],\n"
      "                  buffers: list[bytes-like],\n"
-     "                  use_o_direct: bool = True) -> None\n"
+     "                  use_o_direct: bool = True,\n"
+     "                  checksum: bool = False) -> None\n"
      "\n"
      "Store a batch of blocks, each from its own buffer, to disk. Raises on "
      "first error."},
     {"batch_load_block", batch_load_block, METH_VARARGS,
      "batch_load_block(source_paths: list[str],\n"
      "                 buffers: list[writable bytes-like],\n"
-     "                 use_o_direct: bool = True) -> None\n"
+     "                 use_o_direct: bool = True,\n"
+     "                 checksum: bool = False) -> None\n"
      "\n"
      "Load a batch of blocks from disk into corresponding buffers. "
      "Raises on first error."},

docs/features/kv_offloading_usage.md

@@ -242,9 +242,18 @@ The filesystem tier (`type: "fs"`) writes blocks to a filesystem directory.
 | `n_write_threads` | no | `16` | Write-priority I/O threads (store path). |
 | `enable_kv_events` | no | `false` | Publish `BlockStored` KV events (medium `STORAGE`) for successfully stored blocks. Requires KV cache events to be enabled globally. |
 | `locality` | no | unspecified | `LOCAL` or `REMOTE` relative to the publishing vLLM instance. Included in the tier's KV events only when explicitly configured. |
+| `checksum_blocks` | no | `false` | Record a CRC32C per block and verify it on load (see [Block Checksums](#block-checksums)). Requires user extended attribute support at `root_dir`. |
 
 Each thread group prefers its own queue but pulls from the other when its primary queue is empty, so a write-heavy or read-heavy burst won't leave the off-priority queue waiting. Size the totals to your storage's effective concurrency.
 
+#### Block Checksums
+
+With `checksum_blocks: true`, the tier records each block's CRC32C in the user extended attribute `user.vllm.kv_crc32c` on store and verifies it on load, catching a block whose bytes changed while its size did not (e.g. bit rot). A mismatching block is removed, its load fails, and `vllm:kv_offload_tiering_checksum_failures` is incremented. If `root_dir` does not support user extended attributes, the tier fails at startup.
+
+A block with no recorded checksum still loads unverified, so the option can be enabled on an existing cache. Tools that copy files without extended attributes (`cp` without `-a`, `rsync` without `-X`) produce such blocks. Blocks already in the cache when the option is enabled are not rewritten, so only blocks stored afterwards are verified.
+
+Checksumming adds a CRC32C pass over every block on store and load. The pure-Python I/O fallback, used where the native extension is not built, computes it far more slowly.
+
 #### On-Disk Layout
 
 Under `root_dir`, vLLM creates a subdirectory `<model>_<digest>`, where `<model>` is the model name with `/` replaced by `_` (so HuggingFace IDs like `meta-llama/Llama-3-8B` don't nest), and `<digest>` is a short SHA256 prefix derived from the run configuration (model, block size, parallelism, dtype, etc.). Runs with the same configuration share the same subdirectory; runs with different configurations live side-by-side under the same `root_dir` without colliding.