lmcache--lmcache/docs/design/v1/distributed/serde

lmcache.v1.distributed.serde — Serialization / Deserialization Package

Scope

This package holds the generic serde primitives used by the distributed storage stack. It does not know anything about L2 adapters or controllers — it just defines:

• A pair of sync interfaces users implement (Serializer, Deserializer) for the actual byte transform.
• An async interface (SerdeProcessor) with the same submit → eventfd → query shape as an L2 adapter, so downstream consumers can poll it uniformly.
• A default implementation (AsyncSerdeProcessor) that turns any pair of sync implementations into the async interface by running them in a thread pool and signaling an eventfd on completion.
• A factory / registration mechanism so adapters can reference a serde by name ({"type": "fp8", ...} in JSON config).
• One built-in serde: fp8 quantization.

How the async interface is actually plugged into the L2 path lives in docs/design/v1/distributed/l2_adapters/serde_wrapper.md — the wrapper is the sole consumer of SerdeProcessor's event fds.

Module Layout

lmcache/v1/distributed/serde/
  base.py             # Serializer, Deserializer (sync ABCs)
                      # SerdeProcessor (async ABC)
                      # SerdeConfig, SerdeTaskId
  async_processor.py  # AsyncSerdeProcessor (thread-pool + eventfd wrapper)
  factory.py          # register_serde_factory / create_serde_processor
  fp8.py              # Fp8QuantizationSerializer / Deserializer
  multi.py            # MultiSerializer / MultiDeserializer (tuple-shaped
                      # extension; see "Multi-output extension" below)
  utils.py            # serialized_layout_desc, make_temp_key

Two-Layer Interface

                 ┌────────────────────────────────┐
 user writes →   │  Serializer / Deserializer     │  sync transform
                 │  (pure: src MemoryObj → dst)   │
                 └──────────────┬─────────────────┘
                                │ wrapped by AsyncSerdeProcessor
                                ▼
                 ┌────────────────────────────────┐
 wrapper uses →  │  SerdeProcessor                │  async:
                 │    submit_serialize(...) → id  │   submit / eventfd /
                 │    query_serialize_result(id)  │   query
                 │    get_serialize_event_fd()    │
                 │    (plus deserialize pair)     │
                 └────────────────────────────────┘

• Sync layer is where the user cares. Pure Python (or torch) code, no threads, no fds. Two abstract methods: serialize(src, dst) and estimate_serialized_size(layout_desc).
• Async layer is what the SerdeL2AdapterWrapper talks to. It owns two eventfds (one for serialize, one for deserialize) that must be distinct, and queues completed tasks in a dict the wrapper drains.

AsyncSerdeProcessor is the default and typically only implementation of the async layer — most custom serdes only need to provide the sync classes and register a factory.

Contracts

Serializer.serialize(src, dst) -> int

• src is a MemoryObj holding KV data (read-locked by the caller).
• dst is a MemoryObj byte buffer (write-locked by the caller), sized ≥ estimate_serialized_size(layout_of_src).
• Must return the number of bytes actually written to dst.
• Must be deterministic given the same src — the wrapper relies on the serialize step being reproducible across retries.

Serializer.estimate_serialized_size(layout_desc) -> int

• Called once per batch to size the temp buffer before any work.
• Must be an upper bound on the actual serialized output. Include any safety margin inside this method (the fp8 serializer returns 1.5 × num_elements for exactly this reason).
• Must only depend on layout_desc (shapes + dtypes). The wrapper uses the first object's layout to size temps for the whole batch, so a data-dependent estimate would break all-or-nothing allocation.

Deserializer.deserialize(src, dst) -> None

• src is a byte-buffer MemoryObj filled by L2 load.
• dst is a KV-shaped MemoryObj (write-locked), already the correct shape and dtype.
• No return value — the caller observes completion via the async layer's event fd.

SerdeProcessor (async)

• submit_serialize(src_objs, dst_objs) → SerdeTaskId must be non-blocking. The actual transform runs asynchronously.
• query_serialize_result(task_id) → bool | None is non-idempotent: it returns a non-None value exactly once per task id. None means the task is still in flight.
• get_serialize_event_fd() returns an eventfd signaled once per completed serialize task; distinct from the deserialize fd.
• The deserialize side has the identical shape.
• close() must release both event fds and any worker threads.

AsyncSerdeProcessor

Wraps any (Serializer, Deserializer) pair. Internal design:

• One ThreadPoolExecutor(max_workers=N) runs both serialize and deserialize tasks; they're independent, so the pool is shared.
• One task-id counter under a single lock; two _completed_* dicts (one per direction) protected by the same lock.
• Two os.eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC) file descriptors — one per direction, signaled after the result is written to the completion dict.

Why a pool, not an asyncio executor: the CPU-bound fp8 / encryption transforms release the GIL under torch / native calls, so a real thread pool is useful. N=1 is a safe default (one in-flight transform at a time), and the fp8 factory accepts max_workers to bump it.

Factory / Registration

from lmcache.v1.distributed.serde import (
    AsyncSerdeProcessor, Deserializer, Serializer,
    register_serde_factory,
)

def _create_my_serde(kwargs: dict[str, object]) -> SerdeProcessor:
    return AsyncSerdeProcessor(MySerializer(...), MyDeserializer(...))

register_serde_factory("mine", _create_my_serde)

The factory receives the type-specific kwargs from the JSON config (everything except "type"). The registry is process-global and rejects duplicate names, matching the pattern already used by register_l2_adapter_type.

The factory is called exactly once per SerdeL2AdapterWrapper construction — each wrapped adapter gets its own SerdeProcessor instance.

Built-in fp8

Fp8QuantizationSerializer / Fp8QuantizationDeserializer:

• Cast each element to torch.float8_e4m3fn (default) or torch.float8_e5m2, reinterpret the bytes as uint8, and copy into the temp buffer.
• Deserialize reinterprets the uint8 bytes as the chosen fp8 dtype, reshapes back to the original KV shape, and casts to the destination tensor's dtype.
• estimate_serialized_size returns int(total_elements × 1.5) — the exact fp8 size is num_elements × 1 byte, and the 1.5× headroom absorbs future format changes or alignment padding.

Extension Guide

Most custom serdes only need:

1. Two classes implementing Serializer / Deserializer.
2. A factory function registered at import time.
3. A JSON serde sub-dict on the adapter config: {"type": "mine", ...}.

Everything else — temp buffer allocation, eventfd plumbing, lock / lifecycle transitions, all-or-nothing failure handling — is provided by AsyncSerdeProcessor and the wrapper. Users never need to touch controllers, eventfds, or L1 locks.

Multi-output extension

The Serializer / Deserializer classes above operate on one typed tensor at each endpoint: one tensor in, one byte buffer out on serialize, and the reverse on deserialize. That shape works when K and V share a single data type -- the serde just sees them as one combined tensor. It does not work in two cases:

• K and V at different data types. A typed tensor has one dtype, so a serde that wants K at one dtype (e.g. fp16/bf16) and V at another (e.g. FP8) cannot carry both.
• One side absent. For tier-split placements where K or V is held outside this serde's data path -- e.g. K kept in L1 (CPU pinned host memory) while V flows to L2 (durable storage) -- the serialize input has no tensor for the absent slot and the deserialize output has no destination for it.

multi.py defines the additive contract for this case:

• MemoryObjGroup = Tuple[Optional[MemoryObj], ...] is a fixed-length tuple of optional MemoryObjs.
• LayoutDescGroup = Tuple[Optional[MemoryLayoutDesc], ...] is the parallel layout-descriptor tuple used by size estimators.
• MultiSerializer.serialize(src: MemoryObjGroup, dst: MemoryObj) takes a group whose length equals MultiSerializer.group_size.
• MultiDeserializer.deserialize(src: MemoryObj, dst: MemoryObjGroup) produces a group whose length equals MultiDeserializer.group_size.
• single_to_multi_serializer(s) and single_to_multi_deserializer(d) adapt an existing single-tensor pair to the tuple interface as a length-1 group; the adapter is layout-equivalent (same on-the-wire bytes as a direct call).

The single-tensor Serializer / Deserializer ABCs and all their existing callers — AsyncSerdeProcessor, the factory registry, the L2 adapter wrapper, the built-in fp8 serde — are unchanged. A serde implementation that needs multiple tensors at an endpoint implements MultiSerializer / MultiDeserializer instead of (or in addition to) the single-tensor ABCs. The async wiring around the multi interface — a tuple-aware AsyncSerdeProcessor analog and a tuple-aware submit_* shape on the wrapper — is added in a follow-up once a concrete multi-output serde lands.

Per-slot semantics

Implementations MUST document, at minimum:

1. The fixed value of group_size.
2. The semantic carried by each slot (e.g., slot 0 = K, slot 1 = V).
3. Which slots are required and which may be None. A slot that may be None MUST be tolerated by estimate_serialized_size with a None layout descriptor at the same index.

None semantics, mirroring serialize input vs deserialize output:

• Serialize input. A None slot means the caller is not supplying that tensor; e.g., a V-only write where K is held outside this serde's data path. The implementation MUST raise ValueError when a required slot is None.
• Deserialize output. A None slot means the caller does not want that tensor materialized; e.g., a V-only read where K is sourced elsewhere. The implementation MUST NOT touch the missing slot.

Single-element bridge

A length-1 MemoryObjGroup is the trivial bridge to the existing single-tensor API. single_to_multi_serializer(inner) wraps an existing Serializer so callers that work in groups can invoke it uniformly:

from lmcache.v1.distributed.serde import (
    single_to_multi_serializer,
    single_to_multi_deserializer,
)

multi_s = single_to_multi_serializer(existing_serializer)
multi_d = single_to_multi_deserializer(existing_deserializer)
n = multi_s.serialize((src,), dst_buffer)         # length-1 group
multi_d.deserialize(src_buffer, (dst,))           # length-1 group

The wrapper rejects non-unit groups with ValueError, rejects a None src slot (single-tensor serializers do not admit absence), and treats a None dst slot as a deliberate skip rather than an error. On-the-wire bytes are byte-for-byte identical to a direct call against the underlying single-tensor serde.