Skip to content
117 changes: 106 additions & 11 deletions csrc/storage_backends/fs/connector.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,68 @@ std::string FSConnector::key_to_filename(const std::string& key) {
return result;
}

std::string FSConnector::hex_encode(const std::string& value) {
static constexpr char HEX[] = "0123456789abcdef";
std::string encoded;
encoded.resize(value.size() * 2);
for (size_t index = 0; index < value.size(); ++index) {
const auto byte = static_cast<unsigned char>(value[index]);
encoded[2 * index] = HEX[byte >> 4];
encoded[2 * index + 1] = HEX[byte & 0x0f];
}
return encoded;
}

std::filesystem::path FSConnector::key_to_relative_path(
const std::string& key) {
const std::string legacy_filename = key_to_filename(key);
if (legacy_filename.size() <= LEGACY_FILENAME_MAX_BYTES) {
return legacy_filename;
}

std::vector<std::string> parts;
size_t start = 0;
for (size_t pos = 0; pos <= key.size(); ++pos) {
if (pos == key.size() || key[pos] == KEY_SEP) {
parts.emplace_back(key.substr(start, pos - start));
start = pos + 1;
}
}
if (parts.size() != 4 && parts.size() != 5) {
throw std::runtime_error(
"FSConnector: oversized keys require the current four- or five-field "
"ObjectKey encoding: " +
key);
}

const std::string model_hex = hex_encode(parts[0]);
const std::string salt_hex =
parts.size() == 5 ? hex_encode(parts[4]) : std::string();
const size_t model_count =
(model_hex.size() + ENCODED_COMPONENT_MAX_CHARS - 1) /
ENCODED_COMPONENT_MAX_CHARS;
const size_t salt_count =
(salt_hex.size() + ENCODED_COMPONENT_MAX_CHARS - 1) /
ENCODED_COMPONENT_MAX_CHARS;

std::filesystem::path relative_path = BOUNDED_PATH_VERSION;
relative_path /= "m" + std::to_string(model_count);
for (size_t offset = 0; offset < model_hex.size();
offset += ENCODED_COMPONENT_MAX_CHARS) {
relative_path /= model_hex.substr(offset, ENCODED_COMPONENT_MAX_CHARS);
}
relative_path /= "s" + std::to_string(salt_count);
for (size_t offset = 0; offset < salt_hex.size();
offset += ENCODED_COMPONENT_MAX_CHARS) {
relative_path /= salt_hex.substr(offset, ENCODED_COMPONENT_MAX_CHARS);
}

std::string leaf = "0x" + parts[1] + std::string(1, KEY_SEP) + parts[2] +
std::string(1, KEY_SEP) + parts[3] + FILE_EXT;
relative_path /= leaf;
return relative_path;
}

// ---------------------------------------------------------------
// read/write helpers
// ---------------------------------------------------------------
Expand Down Expand Up @@ -166,6 +228,15 @@ FSConnector::FSConnector(std::string base_path, int num_workers,
// Create base directory
std::filesystem::create_directories(base_path_);

errno = 0;
const long name_max = ::pathconf(base_path_.c_str(), _PC_NAME_MAX);
if ((name_max < 0 && errno != 0) ||
(name_max >= 0 &&
static_cast<size_t>(name_max) < LEGACY_FILENAME_MAX_BYTES)) {
throw std::runtime_error(
"FSConnector requires a filesystem NAME_MAX of at least 255 bytes");
}

// Create tmp directory if configured
if (!relative_tmp_dir_.empty()) {
auto tmp_path = std::filesystem::path(base_path_) / relative_tmp_dir_;
Expand Down Expand Up @@ -199,8 +270,18 @@ WorkerFSConn FSConnector::create_connection() {

void FSConnector::do_single_get(WorkerFSConn& conn, const std::string& key,
void* buf, size_t len, size_t chunk_size) {
std::string filename = key_to_filename(key);
auto file_path = conn.base_path / filename;
const auto relative_path = key_to_relative_path(key);
auto file_path = conn.base_path / relative_path;
if (relative_path.has_parent_path()) {
std::error_code exists_ec;
if (!std::filesystem::exists(file_path, exists_ec)) {
const auto legacy_path = conn.base_path / key_to_filename(key);
exists_ec.clear();
if (std::filesystem::exists(legacy_path, exists_ec)) {
file_path = legacy_path;
}
}
}

int flags = O_RDONLY;
bool do_odirect = conn.use_odirect &&
Expand Down Expand Up @@ -247,13 +328,19 @@ void FSConnector::do_single_get(WorkerFSConn& conn, const std::string& key,
void FSConnector::do_single_set(WorkerFSConn& conn, const std::string& key,
const void* buf, size_t len,
size_t chunk_size) {
std::string filename = key_to_filename(key);
auto file_path = conn.base_path / filename;
const auto relative_path = key_to_relative_path(key);
auto file_path = conn.base_path / relative_path;

// Skip if already stored on disk
if (std::filesystem::exists(file_path)) {
return;
}
if (relative_path.has_parent_path()) {
std::error_code exists_ec;
const auto legacy_path = conn.base_path / key_to_filename(key);
if (std::filesystem::exists(legacy_path, exists_ec)) return;
std::filesystem::create_directories(file_path.parent_path());
}

const auto tmp_dir =
conn.tmp_dir.empty() ? file_path.parent_path() : conn.tmp_dir;
Expand All @@ -267,7 +354,7 @@ void FSConnector::do_single_set(WorkerFSConn& conn, const std::string& key,
for (size_t attempt = 0; attempt < 1024; ++attempt) {
const uint64_t id =
next_temp_file_id.fetch_add(1, std::memory_order_relaxed);
tmp_path = tmp_dir / (filename + TMP_EXT + "." +
tmp_path = tmp_dir / (std::string(".lmcache-write") + TMP_EXT + "." +
std::to_string(static_cast<uint64_t>(::getpid())) +
"." + std::to_string(id));
fd = ::open(tmp_path.c_str(), flags, 0644);
Expand Down Expand Up @@ -312,16 +399,24 @@ void FSConnector::do_single_set(WorkerFSConn& conn, const std::string& key,
}

bool FSConnector::do_single_exists(WorkerFSConn& conn, const std::string& key) {
std::string filename = key_to_filename(key);
auto file_path = conn.base_path / filename;
return std::filesystem::exists(file_path);
const auto relative_path = key_to_relative_path(key);
const auto file_path = conn.base_path / relative_path;
std::error_code exists_ec;
if (std::filesystem::exists(file_path, exists_ec)) return true;
if (!relative_path.has_parent_path()) return false;
exists_ec.clear();
return std::filesystem::exists(conn.base_path / key_to_filename(key),
exists_ec);
}

bool FSConnector::do_single_delete(WorkerFSConn& conn, const std::string& key) {
std::string filename = key_to_filename(key);
auto file_path = conn.base_path / filename;
const auto relative_path = key_to_relative_path(key);
auto file_path = conn.base_path / relative_path;
std::error_code ec;
return std::filesystem::remove(file_path, ec);
if (std::filesystem::remove(file_path, ec)) return true;
if (!relative_path.has_parent_path()) return false;
ec.clear();
return std::filesystem::remove(conn.base_path / key_to_filename(key), ec);
}

} // namespace connector
Expand Down
9 changes: 9 additions & 0 deletions csrc/storage_backends/fs/connector.h
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,9 @@ static constexpr char KEY_SEP = '@';
static constexpr const char* PATH_SLASH_REPLACEMENT = "-SEP-";
static constexpr const char* FILE_EXT = ".data";
static constexpr const char* TMP_EXT = ".tmp";
static constexpr const char* BOUNDED_PATH_VERSION = ".lmcache-objects-v1";
static constexpr size_t LEGACY_FILENAME_MAX_BYTES = 255;
static constexpr size_t ENCODED_COMPONENT_MAX_CHARS = 200;

// Per-worker connection state for the FS connector. O_DIRECT is enabled per
// request only when both the transfer length and caller-owned buffer address
Expand Down Expand Up @@ -69,6 +72,12 @@ class FSConnector : public ConnectorBase<WorkerFSConn> {
// key or a current unsalted key.
static std::string key_to_filename(const std::string& key);

// Preserve the flat filename when it fits NAME_MAX. Oversized keys use a
// reversible hierarchy whose individual components remain bounded.
static std::filesystem::path key_to_relative_path(const std::string& key);

static std::string hex_encode(const std::string& value);

static std::string replace_all(const std::string& str,
const std::string& from,
const std::string& to);
Expand Down
3 changes: 3 additions & 0 deletions docs/design/v1/distributed/l2_adapters/l2_per_user_quota.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,9 @@ This document describes the per-user quota mechanism for L2 adapters: how
per-user storage limits are enforced, how user identity propagates through the
system, and what changes are needed across the codebase.

This feature is supported only in multiprocess (MP) mode; the deprecated
in-process mode is outside its scope.

## Motivation

Currently, L2 eviction operates on **aggregate** storage usage. A single
Expand Down
131 changes: 120 additions & 11 deletions lmcache/v1/distributed/l2_adapters/fs_l2_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,9 @@
File-system based L2 adapter using aiofiles for async I/O.

Stores KV cache objects as raw tensor bytes on disk (no metadata
header). Each ObjectKey maps to a separate ``.data`` file whose
name encodes all key fields so it can be reversed on startup.
header). Each ObjectKey maps to a separate ``.data`` file. Names that fit the
filesystem component limit use the flat layout. Oversized names use bounded,
reversible path components so restart inventory retains the complete key.
"""

# Future
Expand Down Expand Up @@ -56,6 +57,10 @@
# csrc/storage_backends/fs/connector.cpp.
_PATH_SLASH_REPLACEMENT = "-SEP-"
_FILE_EXT = ".data"
# Keep ObjectKey-to-filename mapping independent of local filesystem settings.
_LEGACY_FILENAME_MAX_BYTES = 255
_BOUNDED_PATH_VERSION = ".lmcache-objects-v1"
_ENCODED_COMPONENT_MAX_CHARS = 200


def _write_all(fd: int, buf: bytes | bytearray | memoryview) -> None:
Expand Down Expand Up @@ -157,7 +162,7 @@ def _object_key_to_filename(key: ObjectKey) -> str:
def _filename_to_object_key(
filename: str,
) -> Optional[ObjectKey]:
"""Reverse ``_object_key_to_filename``.
"""Reverse legacy filenames produced by ``_object_key_to_filename``.

Accepts both the 4-field unsalted shape and the 5-field salted
shape (trailing ``cache_salt``). Returns ``None`` for anything
Expand Down Expand Up @@ -197,6 +202,90 @@ def _filename_to_object_key(
return None


def _hex_path_components(value: str) -> list[str]:
"""Encode a key field into bounded, path-safe components."""
encoded = value.encode("utf-8", errors="surrogatepass").hex()
return [
encoded[offset : offset + _ENCODED_COMPONENT_MAX_CHARS]
for offset in range(0, len(encoded), _ENCODED_COMPONENT_MAX_CHARS)
]


def _object_key_to_bounded_relative_path(key: ObjectKey) -> Path:
"""Map an oversized ObjectKey to a reversible bounded relative path.

Model identity and tenant salt are preserved exactly so restart inventory
can reconstruct capacity and per-tenant accounting without a sidecar.
"""
model_parts = _hex_path_components(key.model_name)
salt_parts = _hex_path_components(key.cache_salt)
leaf = (
f"{key.kv_rank:#010x}{_KEY_SEP}{key.object_group_id:x}"
f"{_KEY_SEP}{key.chunk_hash.hex()}{_FILE_EXT}"
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
)
return Path(
_BOUNDED_PATH_VERSION,
f"m{len(model_parts)}",
*model_parts,
f"s{len(salt_parts)}",
*salt_parts,
leaf,
)


def _object_key_to_relative_path(key: ObjectKey) -> Path:
"""Return a stable flat or bounded storage path for an ObjectKey."""
legacy_filename = _object_key_to_filename(key)
canonical_bytes = legacy_filename.encode("utf-8", errors="surrogatepass")
if len(canonical_bytes) <= _LEGACY_FILENAME_MAX_BYTES:
return Path(legacy_filename)
return _object_key_to_bounded_relative_path(key)


def _bounded_relative_path_to_object_key(
relative_path: Path,
) -> Optional[ObjectKey]:
"""Decode the reversible bounded path shape into its ObjectKey."""
parts = relative_path.parts
if len(parts) < 4 or parts[0] != _BOUNDED_PATH_VERSION:
return None
try:
model_count = int(parts[1][1:]) if parts[1].startswith("m") else -1
model_start = 2
salt_marker = model_start + model_count
if model_count < 0 or salt_marker >= len(parts):
return None
salt_header = parts[salt_marker]
salt_count = int(salt_header[1:]) if salt_header.startswith("s") else -1
salt_start = salt_marker + 1
leaf_index = salt_start + salt_count
if salt_count < 0 or leaf_index != len(parts) - 1:
return None

model_name = bytes.fromhex("".join(parts[model_start:salt_marker])).decode(
"utf-8", errors="surrogatepass"
)
cache_salt = bytes.fromhex("".join(parts[salt_start:leaf_index])).decode(
"utf-8", errors="surrogatepass"
)
leaf = parts[leaf_index]
if not leaf.endswith(_FILE_EXT):
return None
key_parts = leaf[: -len(_FILE_EXT)].split(_KEY_SEP)
if len(key_parts) != 3:
return None
kv_rank_str, object_group_str, chunk_hash_hex = key_parts
return ObjectKey(
chunk_hash=bytes.fromhex(chunk_hash_hex),
model_name=model_name,
kv_rank=int(kv_rank_str, 16),
object_group_id=int(object_group_str, 16),
cache_salt=cache_salt,
)
except (UnicodeDecodeError, ValueError):
return None


class FSL2AdapterConfig(L2AdapterConfigBase):
"""
Config for the filesystem-backed L2 adapter.
Expand Down Expand Up @@ -277,9 +366,9 @@ class FSL2Adapter(L2AdapterInterface):
"""
File-system backed L2 adapter with async I/O via *aiofiles*.

Each file stores **only** the raw tensor bytes (no metadata
header), which gives maximum I/O throughput. The file name
itself encodes the full ``ObjectKey`` so it is reversible.
Each file stores **only** the raw tensor bytes (no metadata header), which
gives maximum I/O throughput. Representable file names encode the complete
``ObjectKey``; oversized names use a reversible bounded path.

Thread safety is ensured via a lock for shared bookkeeping
and an asyncio event loop running on a dedicated daemon
Expand Down Expand Up @@ -310,6 +399,22 @@ def __init__(self, config: FSL2AdapterConfig):
parents=False, exist_ok=True
)

try:
self._name_max = os.pathconf(self._base_path, "PC_NAME_MAX")
if self._relative_tmp_dir is not None:
tmp_path = self._base_path / self._relative_tmp_dir
tmp_name_max = os.pathconf(tmp_path, "PC_NAME_MAX")
self._name_max = min(self._name_max, tmp_name_max)
except (OSError, ValueError) as exc:
raise RuntimeError(
f"Failed to determine PC_NAME_MAX for FS L2 adapter: {exc}"
) from exc
if self._name_max < _LEGACY_FILENAME_MAX_BYTES:
raise ValueError(
"FS L2 adapter requires PC_NAME_MAX >= "
f"{_LEGACY_FILENAME_MAX_BYTES}, got {self._name_max}"
)

# I/O tuning options aligned with FSConnector
self._read_ahead_size = config.read_ahead_size
self._use_odirect = config.use_odirect
Expand Down Expand Up @@ -533,7 +638,7 @@ def _get_next_task_id(self) -> L2TaskId:
return tid

def _key_to_path(self, key: ObjectKey) -> Path:
return self._base_path / _object_key_to_filename(key)
return self._base_path / _object_key_to_relative_path(key)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved

async def _key_exists_on_disk(
self,
Expand All @@ -556,13 +661,17 @@ def _key_to_file_and_tmp_path(self, key: ObjectKey) -> tuple[Path, Path]:
``FSConnector._get_file_and_tmp_path``). Otherwise a
``.tmp`` suffix is used.
"""
fname = _object_key_to_filename(key)
final = self._base_path / fname
relative_path = _object_key_to_relative_path(key)
final = self._base_path / relative_path
final.parent.mkdir(parents=True, exist_ok=True)
if self._relative_tmp_dir is not None:
tmp_dir = self._base_path / self._relative_tmp_dir
else:
tmp_dir = self._base_path
tmp = tmp_dir / f"{fname}.tmp"
tmp_dir = final.parent
# The caller appends a PID and UUID. Keep the temporary basename
# independent of the final key so a valid 255-byte legacy name cannot
# overflow the same filesystem's component limit during publication.
tmp = tmp_dir / ".lmcache-write.tmp"
return final, tmp

# ---- O_DIRECT helpers -----------------------------------------------
Expand Down
Loading
Loading