lmcache--lmcache/docs/design/v1/distributed/l2_adapters/serde_wrapper.md

SerdeL2AdapterWrapper — Transparent Serde via Adapter Composition

Scope

Describes how serialization / deserialization is integrated into the L2 path. The serde package itself (docs/design/v1/distributed/serde/README.md) defines the generic Serializer / Deserializer / SerdeProcessor interfaces and the fp8 built-in. This doc is about the adapter that stitches serde into the distributed storage pipeline.

Design Summary

SerdeL2AdapterWrapper implements L2AdapterInterface by composing an inner L2 adapter with a SerdeProcessor and an L1Manager. The caller sees the wrapper's public API only; the wrapper's internal thread is the sole consumer of the inner adapter's and serde's event fds.

Step numbers below show call ordering. Solid ─► arrows are direct synchronous calls; dotted ╌► arrows are eventfd wakeups consumed by the wrapper's internal thread.

Store path

   caller                                                          
     │                                                             
     │ (1) submit_store_task(keys, objs)                           
     ▼                                                             
   ┌─────────────────┐  (2) reserve_write(tmp)   ┌──────────────┐  
   │  wrapper (API)  │ ─────────────────────────►│  L1Manager   │  
   │                 │ ◄────── tmp_objs ─────────│              │  
   └────────┬────────┘                           └──────────────┘  
            │ (3) submit_serialize(objs, tmp_objs)                 
            ▼                                                      
   ┌─────────────────┐                                             
   │ SerdeProcessor  │   transforms objs → tmp_objs                
   └────────┬────────┘                                             
            ╎ (4) serialize_efd                                    
            ▼                                                      
   ┌─────────────────┐  (5) inner.submit_store_   ┌──────────────┐ 
   │ wrapper thread  │      task(keys, tmp_objs)  │   inner L2   │ 
   │   (_loop)       │ ─────────────────────────► │   adapter    │ 
   └────────┬────────┘                            └──────┬───────┘ 
            ▲                                            │         
            ╎ (6) inner.store_efd ◄──────────────────────┘         
            │                                                      
   ┌────────┴────────┐                                             
   │ wrapper thread  │  (7) finish_read(tmp_objs)  → auto-delete   
   │                 │      signal wrapper.store_efd               
   └────────┬────────┘                                             
            │                                                      
            │ (8) pop_completed_store_tasks() →                   
            │     {wrapped_id: L2StoreResult(success=True,         
            │                                bytes_transferred=N)}
            ▼                                                      
          caller                                                   

Load path

   caller                                                          
     │                                                             
     │ (1) submit_load_task(keys, dst_objs)                        
     ▼                                                             
   ┌─────────────────┐  (2) reserve_write(tmp)   ┌──────────────┐  
   │  wrapper (API)  │ ─────────────────────────►│  L1Manager   │  
   │                 │ ◄────── tmp_objs ─────────│              │  
   └────────┬────────┘                           └──────────────┘  
            │ (3) inner.submit_load_task(keys, tmp_objs)           
            ▼                                                      
   ┌─────────────────┐                                             
   │  inner L2       │   loads serialized bytes into tmp_objs      
   │  adapter        │                                             
   └────────┬────────┘                                             
            ╎ (4) inner.load_efd                                   
            ▼                                                      
   ┌─────────────────┐  (5) submit_deserialize    ┌──────────────┐ 
   │ wrapper thread  │      (tmp_objs, dst_objs)  │   Serde      │ 
   │   (_loop)       │ ─────────────────────────► │  Processor   │ 
   └────────┬────────┘                            └──────┬───────┘ 
            ▲                                            │         
            ╎ (6) deserialize_efd ◄──────────────────────┘         
            │                                                      
   ┌────────┴────────┐                                             
   │ wrapper thread  │  (7) finish_write + delete(tmp_objs)        
   │                 │      signal wrapper.load_efd                
   └────────┬────────┘                                             
            │                                                      
            │ (8) query_load_result() → per-key bitmap             
            ▼                                                      
          caller                                                   

Store Path

sequenceDiagram
    participant Controller as StoreController
    participant W as SerdeL2AdapterWrapper
    participant Thread as wrapper internal thread
    participant Serde as SerdeProcessor
    participant L1 as L1Manager
    participant Inner as inner L2 adapter

    Controller->>W: submit_store_task(keys, objs)
    W->>L1: reserve_write(temp_keys, byte layout)
    L1-->>W: temp_objs (write-locked)
    W->>Serde: submit_serialize(objs, temp_objs) → serde_id
    W-->>Controller: wrapped_id

    Note over Serde,Thread: serde thread pool transforms<br/>objs → temp_objs, signals fd

    Serde-->>Thread: serialize_efd fires
    Thread->>Serde: query_serialize_result(serde_id)
    Serde-->>Thread: success
    Thread->>L1: finish_write_and_reserve_read(temp_keys)
    Thread->>Inner: submit_store_task(keys, temp_objs) → inner_id

    Note over Inner: inner stores serialized bytes,<br/>signals its store_efd

    Inner-->>Thread: store_efd fires
    Thread->>Inner: pop_completed_store_tasks()
    Inner-->>Thread: {inner_id: L2StoreResult(ok=True, bytes=N)}
    Thread->>L1: finish_read(temp_keys)   %% auto-deletes temps
    Thread->>W: _finalize_store(wrapped_id, inner_result)
    W-->>Controller: store_efd fires

    Controller->>W: pop_completed_store_tasks()
    W-->>Controller: {wrapped_id: L2StoreResult(ok=True, bytes=N)}

Load Path

sequenceDiagram
    participant Controller as PrefetchController
    participant W as SerdeL2AdapterWrapper
    participant Thread as wrapper internal thread
    participant Inner as inner L2 adapter
    participant Serde as SerdeProcessor
    participant L1 as L1Manager

    Controller->>W: submit_load_task(keys, dst_objs)
    W->>L1: reserve_write(temp_keys, byte layout)
    L1-->>W: temp_objs (write-locked)
    W->>Inner: submit_load_task(keys, temp_objs) → inner_id
    W-->>Controller: wrapped_id

    Note over Inner: inner loads serialized bytes<br/>into temp_objs, signals load_efd

    Inner-->>Thread: load_efd fires
    Thread->>Inner: query_load_result(inner_id)
    Inner-->>Thread: bitmap (per-key success)

    alt all-zeros bitmap
        Thread->>L1: finish_write + delete temp_keys
        Thread->>W: _finalize_load(wrapped_id, bitmap)
    else some keys loaded
        Thread->>Serde: submit_deserialize(temp_objs, dst_objs) → serde_id
        Note over Serde: serde thread transforms<br/>temp_objs → dst_objs
        Serde-->>Thread: deserialize_efd fires
        Thread->>Serde: query_deserialize_result(serde_id)
        alt success
            Thread->>L1: finish_write + delete temp_keys
            Thread->>W: _finalize_load(wrapped_id, bitmap)
        else failure
            Thread->>L1: finish_write + delete temp_keys
            Thread->>W: _finalize_load(wrapped_id, Bitmap(0, len))
            Note right of W: zero bitmap ⇒ controller<br/>treats keys as load failures
        end
    end

    W-->>Controller: load_efd fires
    Controller->>W: query_load_result(wrapped_id)
    W-->>Controller: final bitmap

Lookup / Unlock / Eviction

These paths don't involve any transform, so the wrapper delegates directly to the inner adapter — including the lookup event fd itself, to avoid an unnecessary thread hop per lookup.

API	Behavior
get_lookup_and_lock_event_fd	Returns the inner adapter's fd (pass-through)
submit_lookup_and_lock_task / query_lookup_and_lock_result	Delegated directly
submit_unlock	Delegated directly
delete / get_usage / supports_global_eviction	Delegated directly
register_listener	Registers on the inner adapter (listeners track real storage state)

The wrapper does not maintain its own byte accounting — it reports whatever the inner adapter reports. This keeps L2EvictionController and per-cache_salt quota logic unchanged: they see the inner adapter's byte totals through the wrapper's get_usage().

Temp Buffer Lifecycle

Temp byte buffers are the only new L1 state the wrapper introduces. They exist entirely within the wrapper's knowledge — the caller never sees the temp keys.

Store path:

1. reserve_write(temp_keys, is_temporary=True, layout=ser_layout, mode="new") — temps are write-locked and marked temporary so finish_read will auto-delete them later.
2. Serialize runs, filling temps.
3. On success: finish_write_and_reserve_read(temp_keys) — temps become read-locked so inner.submit_store_task can safely read them.
4. Inner store completes → finish_read(temp_keys) — since is_temporary=True, finish_read also deletes them.
5. On serialize or inner failure: finish_write(temp_keys) + delete(temp_keys) while temps are still write-locked.

Load path:

1. reserve_write(temp_keys, is_temporary=True, layout=ser_layout, mode="new") — same as store, temps write-locked.
2. inner.submit_load_task(keys, temp_objs) — inner loads serialized bytes into temps.
3. Inner completes → submit_deserialize(temp_objs, dst_objs) — note temps stay write-locked (the wrapper owns them; only the wrapper reads them during deserialize).
4. Deserialize completes → finish_write(temp_keys) + delete(temp_keys) regardless of deserialize success.

Failure Policy: All-or-Nothing per Submit

If any key's temp allocation fails or submit_serialize / submit_deserialize / inner.submit_* raises, the whole wrapped task fails:

• Store task: pop_completed_store_tasks() returns {wrapped_id: L2StoreResult(success=False, bytes_transferred=0)}.
• Load task: query_load_result(wrapped_id) returns an all-zeros Bitmap(len(keys)).

This preserves the coarse-grained success semantic of L2AdapterInterface — store is task-level (the L2StoreResult.is_successful() flag), load is per-key via bitmap. The alternative ("drop the failed keys, succeed with the rest") would silently violate the caller's assumption that every key it passed either succeeded (task-level) or failed (task-level). Keeping the policy coarse is what lets the controllers stay untouched.

Partial load failures inside the inner adapter (some keys in the inner bitmap succeed, some fail) are faithfully preserved: only keys with bitmap.test(i) == True are deserialized, and the wrapper reports the same bitmap to the controller (zeroed if deserialize itself fails).

Homogeneity Invariant

_alloc_temp_buffers sizes temps from objects[0].get_shapes() / get_dtypes() under the assumption that all MemoryObjs in one submit share a single layout. The store controller already shape-groups keys before calling submit_store_task; the prefetch controller uses a single layout_desc per request for all write reservations. The wrapper enforces the invariant with an explicit raise ValueError to catch future regressions.

Threading Model

One background thread per wrapper instance, started in __init__.

• Poll loop over four fds: inner.store_efd, inner.load_efd, serde.serialize_efd, serde.deserialize_efd.
• Single threading.Lock protecting: the task-id counter, the four reverse-lookup dicts, the two completion dicts, and the phase flip on store tasks.
• close() sets a stop flag and joins the thread; the poll timeout (500 ms) bounds shutdown latency.
• _finalize_store / _finalize_load write 1 to the wrapper's own eventfd to wake the upstream controller's poll loop.

The wrapper does not share any state with the inner adapter or the SerdeProcessor beyond the public interface — both are independent thread-safe components, and the wrapper's thread interacts with them only through their documented APIs.

Known Limitations / Follow-ups

• Temp buffer allocation per submit. Every store / load call issues an L1Manager.reserve_write for fresh temp keys. A free-list keyed on (shape, dtype) would skip the allocator on the hot path. Not done in the initial version — correctness first.
• Bookkeeping-dict duplication. Six of the existing L2 adapters maintain their own _next_task_id + lock + completion_dict + eventfd quartet; AsyncSerdeProcessor is a seventh; this wrapper is an eighth. An EventfdTaskQueue base helper could dedupe this across the tree — a separate cleanup PR.
• No wrapper-level metrics. report_status delegates to the inner and adds {"serde_wrapped": True}. Adding per-serde-step latency histograms would need either new metrics from AsyncSerdeProcessor or timing hooks in the wrapper's drains.