SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

perf(data): bound lidar readers and lab session loading

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-07-30 21:52:10 +03:00
parent 7d1a70d8e0
commit e9bbfb9a41
10 changed files with 580 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/web/app.py
+6
View File
@@ -43,6 +43,7 @@ from k1link.web.e30_review_api import build_e30_review_router
from k1link.web.environment_api import build_environment_router
from k1link.web.laboratory_api import build_laboratory_router
from k1link.web.lidar_api import build_lidar_router
from k1link.web.lidar_local_surface_service import K1LocalSurfaceReadService
from k1link.web.map_api import (
MapGatewayConfiguration,
MapGatewayProxy,
@@ -90,6 +91,9 @@ plugin_catalog: DevicePluginCatalog = plugin_environment.catalog
plugin_dispatcher: DevicePluginDispatcher = plugin_environment.dispatcher
session_store = SessionStore(REPOSITORY_ROOT)
session_artifact_gateway = configured_artifact_gateway(session_store.data_dir)
lidar_local_surface_read_service = K1LocalSurfaceReadService(
session_store.data_dir / "lidar-read-cache"
)
session_recording_materializer = SessionRecordingMaterializer(
session_store.data_dir,
exporters=plugin_environment.recording_exporters,
@@ -308,6 +312,7 @@ async def app_lifespan(_: FastAPI) -> AsyncIterator[None]:
with suppress(asyncio.CancelledError):
await reconciler
await asyncio.to_thread(session_recording_preparation_manager.close)
await asyncio.to_thread(lidar_local_surface_read_service.close)
plugin_environment.close()
@@ -493,6 +498,7 @@ app.include_router(
dataset_ground_preview_provider=lambda: (
REPOSITORY_ROOT / ".runtime" / "dataset-gateway" / "ground-comparison.json"
),
local_surface_read_service=lidar_local_surface_read_service,
)
)
app.include_router(
src/k1link/web/lidar_api.py
+25 -78
View File
@@ -9,14 +9,11 @@ from typing import Annotated, Any, Final
from fastapi import APIRouter, HTTPException, Query, Response
from k1link.compute import (
E10LidarFieldSource,
K1LocalSurfaceV1,
LidarFieldReviewV1,
LidarGroundBenchmarkV1,
LidarGroundError,
LidarReplayError,
LidarReplayPackV2,
k1_local_surface_catalog_item,
lidar_field_review_catalog_item,
lidar_ground_benchmark_catalog_item,
lidar_ground_frame_detail,
@@ -33,6 +30,10 @@ from k1link.datasets import (
read_dataset_ground_preview,
read_dataset_native_scan_preview,
)
from k1link.web.lidar_local_surface_service import (
K1LocalSurfaceReadService,
LocalSurfaceSourceUnavailable,
)
LIDAR_CATALOG_SCHEMA: Final = "missioncore.lidar-replay-pack-catalog/v1"
LIDAR_GROUND_CATALOG_SCHEMA: Final = "missioncore.lidar-ground-benchmark-catalog/v1"
@@ -42,7 +43,6 @@ _PACK_ID = re.compile(r"^lidar-replay-pack-[a-f0-9]{64}$")
_BENCHMARK_ID = re.compile(r"^ground-benchmark-[a-f0-9]{64}$")
_FIELD_REVIEW_ID = re.compile(r"^lidar-field-review-[a-f0-9]{64}$")
_LOCAL_SURFACE_ID = re.compile(r"^k1-local-surface-[a-f0-9]{64}$")
_E10_PACK_ID = re.compile(r"^e10-lidar-pack-[a-f0-9]{64}$")
RootProvider = Callable[[], Path | None]
DatasetArtifactProvider = Callable[[], Path | None]
@@ -84,8 +84,10 @@ def build_lidar_router(
dataset_rellis_preview_provider: DatasetArtifactProvider = lambda: None,
dataset_rellis_admission_provider: DatasetArtifactProvider = lambda: None,
dataset_ground_preview_provider: DatasetArtifactProvider = configured_dataset_ground_preview,
local_surface_read_service: K1LocalSurfaceReadService | None = None,
) -> APIRouter:
router = APIRouter(prefix="/api/v1/lidar", tags=["lidar"])
surface_reader = local_surface_read_service or K1LocalSurfaceReadService()
@router.get("/dataset-gateway")
def get_dataset_gateway() -> dict[str, object]:
@@ -531,11 +533,7 @@ def build_lidar_router(
if len(items) >= limit:
break
try:
model = K1LocalSurfaceV1(candidate)
try:
items.append(k1_local_surface_catalog_item(model))
finally:
model.close()
items.append(surface_reader.catalog_item(candidate))
except (LidarGroundError, OSError):
invalid_total += 1
return {
@@ -575,29 +573,12 @@ def build_lidar_router(
detail="K1 local-surface model не найден",
)
try:
model = K1LocalSurfaceV1(model_path)
try:
source_pack_id = model.identity.get("source_pack_id")
if (
not isinstance(source_pack_id, str)
or _E10_PACK_ID.fullmatch(source_pack_id) is None
):
raise LidarGroundError("K1 local-surface source id is invalid")
source_path = source_root / source_pack_id
if not source_path.is_dir():
raise HTTPException(
status_code=404,
detail="Связанный E10 LiDAR source не найден",
)
source = E10LidarFieldSource(source_path)
try:
return model.timeline_detail(source)
finally:
source.close()
finally:
model.close()
except HTTPException:
raise
return surface_reader.timeline_detail(model_path, source_root)
except LocalSurfaceSourceUnavailable as exc:
raise HTTPException(
status_code=404,
detail="Связанный E10 LiDAR source не найден",
) from exc
except (LidarGroundError, OSError) as exc:
raise HTTPException(
status_code=409,
@@ -630,29 +611,12 @@ def build_lidar_router(
detail="K1 local-surface model не найден",
)
try:
model = K1LocalSurfaceV1(model_path)
try:
source_pack_id = model.identity.get("source_pack_id")
if (
not isinstance(source_pack_id, str)
or _E10_PACK_ID.fullmatch(source_pack_id) is None
):
raise LidarGroundError("K1 local-surface source id is invalid")
source_path = source_root / source_pack_id
if not source_path.is_dir():
raise HTTPException(
status_code=404,
detail="Связанный E10 LiDAR source не найден",
)
source = E10LidarFieldSource(source_path)
try:
return model.review_detail(source)
finally:
source.close()
finally:
model.close()
except HTTPException:
raise
return surface_reader.review_detail(model_path, source_root)
except LocalSurfaceSourceUnavailable as exc:
raise HTTPException(
status_code=404,
detail="Связанный E10 LiDAR source не найден",
) from exc
except (LidarGroundError, OSError) as exc:
raise HTTPException(
status_code=409,
@@ -688,34 +652,17 @@ def build_lidar_router(
detail="K1 local-surface model не найден",
)
try:
model = K1LocalSurfaceV1(model_path)
try:
source_pack_id = model.identity.get("source_pack_id")
if (
not isinstance(source_pack_id, str)
or _E10_PACK_ID.fullmatch(source_pack_id) is None
):
raise LidarGroundError("K1 local-surface source id is invalid")
source_path = source_root / source_pack_id
if not source_path.is_dir():
raise HTTPException(
status_code=404,
detail="Связанный E10 LiDAR source не найден",
)
source = E10LidarFieldSource(source_path)
try:
return model.frame_detail(source, frame_index)
finally:
source.close()
finally:
model.close()
return surface_reader.frame_detail(model_path, source_root, frame_index)
except IndexError as exc:
raise HTTPException(
status_code=404,
detail="K1 local-surface frame не найден",
) from exc
except HTTPException:
raise
except LocalSurfaceSourceUnavailable as exc:
raise HTTPException(
status_code=404,
detail="Связанный E10 LiDAR source не найден",
) from exc
except (LidarGroundError, OSError) as exc:
raise HTTPException(
status_code=409,
src/k1link/web/lidar_local_surface_service.py
+346
View File
@@ -0,0 +1,346 @@
"""Bounded, restart-safe readers for immutable E28 laboratory evidence."""
from __future__ import annotations
import json
import os
import re
import stat
import threading
from collections import OrderedDict
from dataclasses import dataclass
from pathlib import Path
from typing import Final
from uuid import uuid4
from k1link.compute.lidar_field_review import E10LidarFieldSource
from k1link.compute.lidar_ground import LidarGroundError
from k1link.compute.lidar_local_surface import (
K1_LOCAL_SURFACE_ARRAYS_NAME,
K1_LOCAL_SURFACE_MANIFEST_NAME,
K1_LOCAL_SURFACE_REPORT_NAME,
K1LocalSurfaceV1,
k1_local_surface_catalog_item,
)
VALIDATION_CACHE_SCHEMA: Final = "missioncore.lidar-read-validation-cache/v1"
E10_LIDAR_ARRAYS_NAME: Final = "lidar-pack.npz"
E10_LIDAR_MANIFEST_NAME: Final = "manifest.json"
DEFAULT_READER_CACHE_ENTRIES: Final = 2
_E10_PACK_ID = re.compile(r"^e10-lidar-pack-[a-f0-9]{64}$")
_MAX_PROOF_BYTES = 64 * 1024
_FileIdentity = tuple[int, int, int, int, int]
_Generation = tuple[tuple[str, _FileIdentity], ...]
class LocalSurfaceSourceUnavailable(LidarGroundError):
"""The immutable model points to a source pack absent from this host."""
@dataclass(slots=True)
class _ModelEntry:
generation: _Generation
reader: K1LocalSurfaceV1
@dataclass(slots=True)
class _SourceEntry:
generation: _Generation
reader: E10LidarFieldSource
class K1LocalSurfaceReadService:
"""Own strict admission, durable validation proofs, and bounded NPZ handles.
A changed generation is always read by the strict compute reader first.
Once admitted, later process starts may restore that exact inode/stat
generation using the private proof without hashing or scanning the large
arrays again. Public read operations remain serialized so parallel LAB
bootstrap requests cannot multiply disk and memory pressure.
"""
def __init__(
self,
cache_root: Path | None = None,
*,
max_entries: int = DEFAULT_READER_CACHE_ENTRIES,
) -> None:
if max_entries < 1:
raise ValueError("LiDAR reader cache must retain at least one entry")
self.cache_root = (
cache_root.expanduser().absolute() if cache_root is not None else None
)
self.max_entries = max_entries
self._lock = threading.RLock()
self._models: OrderedDict[Path, _ModelEntry] = OrderedDict()
self._sources: OrderedDict[Path, _SourceEntry] = OrderedDict()
def close(self) -> None:
with self._lock:
for entry in self._models.values():
entry.reader.close()
for source_entry in self._sources.values():
source_entry.reader.close()
self._models.clear()
self._sources.clear()
def catalog_item(self, model_path: Path) -> dict[str, object]:
with self._lock:
return k1_local_surface_catalog_item(self._model(model_path))
def timeline_detail(
self,
model_path: Path,
source_root: Path,
) -> dict[str, object]:
with self._lock:
model, source = self._bound_readers(model_path, source_root)
return model.timeline_detail(source)
def review_detail(
self,
model_path: Path,
source_root: Path,
) -> dict[str, object]:
with self._lock:
model, source = self._bound_readers(model_path, source_root)
return model.review_detail(source)
def frame_detail(
self,
model_path: Path,
source_root: Path,
frame_index: int,
) -> dict[str, object]:
with self._lock:
model, source = self._bound_readers(model_path, source_root)
return model.frame_detail(source, frame_index)
def _bound_readers(
self,
model_path: Path,
source_root: Path,
) -> tuple[K1LocalSurfaceV1, E10LidarFieldSource]:
model = self._model(model_path)
source_pack_id = model.identity.get("source_pack_id")
if (
not isinstance(source_pack_id, str)
or _E10_PACK_ID.fullmatch(source_pack_id) is None
):
raise LidarGroundError("K1 local-surface source id is invalid")
source_path = source_root / source_pack_id
if not source_path.is_dir():
raise LocalSurfaceSourceUnavailable("linked E10 LiDAR source is unavailable")
return model, self._source(source_path)
def _model(self, path: Path) -> K1LocalSurfaceV1:
root = path.expanduser().absolute()
generation = _generation(
root,
(
K1_LOCAL_SURFACE_MANIFEST_NAME,
K1_LOCAL_SURFACE_REPORT_NAME,
K1_LOCAL_SURFACE_ARRAYS_NAME,
),
)
cached = self._models.get(root)
if cached is not None and cached.generation == generation:
self._models.move_to_end(root)
return cached.reader
if cached is not None:
cached.reader.close()
del self._models[root]
proof_hit = self._proof_matches("models", root.name, generation)
reader = (
K1LocalSurfaceV1._restore_validated_generation(root)
if proof_hit
else K1LocalSurfaceV1(root)
)
try:
stable_generation = _generation(
root,
(
K1_LOCAL_SURFACE_MANIFEST_NAME,
K1_LOCAL_SURFACE_REPORT_NAME,
K1_LOCAL_SURFACE_ARRAYS_NAME,
),
)
if stable_generation != generation:
raise LidarGroundError("K1 local-surface generation changed during admission")
if not proof_hit:
self._publish_proof("models", root.name, stable_generation)
except BaseException:
reader.close()
raise
self._models[root] = _ModelEntry(stable_generation, reader)
self._evict_models()
return reader
def _source(self, path: Path) -> E10LidarFieldSource:
root = path.expanduser().absolute()
generation = _generation(
root,
(E10_LIDAR_MANIFEST_NAME, E10_LIDAR_ARRAYS_NAME),
)
cached = self._sources.get(root)
if cached is not None and cached.generation == generation:
self._sources.move_to_end(root)
return cached.reader
if cached is not None:
cached.reader.close()
del self._sources[root]
proof_hit = self._proof_matches("sources", root.name, generation)
reader = (
E10LidarFieldSource._restore_validated_generation(root)
if proof_hit
else E10LidarFieldSource(root)
)
try:
stable_generation = _generation(
root,
(E10_LIDAR_MANIFEST_NAME, E10_LIDAR_ARRAYS_NAME),
)
if stable_generation != generation:
raise LidarGroundError("E10 LiDAR source generation changed during admission")
if not proof_hit:
self._publish_proof("sources", root.name, stable_generation)
except BaseException:
reader.close()
raise
self._sources[root] = _SourceEntry(stable_generation, reader)
self._evict_sources()
return reader
def _proof_matches(
self,
kind: str,
artifact_id: str,
generation: _Generation,
) -> bool:
if self.cache_root is None:
return False
path = self.cache_root / kind / f"{artifact_id}.json"
try:
metadata = path.lstat()
if (
stat.S_ISLNK(metadata.st_mode)
or not stat.S_ISREG(metadata.st_mode)
or metadata.st_size > _MAX_PROOF_BYTES
):
return False
document: object = json.loads(path.read_text(encoding="utf-8"))
except (FileNotFoundError, OSError, json.JSONDecodeError):
return False
return document == _proof_document(kind, artifact_id, generation)
def _publish_proof(
self,
kind: str,
artifact_id: str,
generation: _Generation,
) -> None:
if self.cache_root is None:
return
root = _private_directory(self.cache_root)
destination_root = _private_directory(root / kind)
destination = destination_root / f"{artifact_id}.json"
temporary = destination_root / f".{artifact_id}.{uuid4().hex}.tmp"
payload = json.dumps(
_proof_document(kind, artifact_id, generation),
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
).encode("utf-8")
descriptor = os.open(
temporary,
os.O_WRONLY | os.O_CREAT | os.O_EXCL,
0o600,
)
try:
with os.fdopen(descriptor, "wb") as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, destination)
finally:
temporary.unlink(missing_ok=True)
def _evict_models(self) -> None:
while len(self._models) > self.max_entries:
_, entry = self._models.popitem(last=False)
entry.reader.close()
def _evict_sources(self) -> None:
while len(self._sources) > self.max_entries:
_, entry = self._sources.popitem(last=False)
entry.reader.close()
def _proof_document(
kind: str,
artifact_id: str,
generation: _Generation,
) -> dict[str, object]:
return {
"schema_version": VALIDATION_CACHE_SCHEMA,
"kind": kind,
"artifact_id": artifact_id,
"files": [
{
"name": name,
"device": identity[0],
"inode": identity[1],
"byte_length": identity[2],
"mtime_ns": identity[3],
"ctime_ns": identity[4],
}
for name, identity in generation
],
}
def _generation(root: Path, names: tuple[str, ...]) -> _Generation:
try:
root_metadata = root.lstat()
except OSError as exc:
raise LidarGroundError("LiDAR evidence root is unavailable") from exc
if stat.S_ISLNK(root_metadata.st_mode) or not stat.S_ISDIR(root_metadata.st_mode):
raise LidarGroundError("LiDAR evidence root is unsafe")
return tuple((name, _regular_file_identity(root / name)) for name in names)
def _regular_file_identity(path: Path) -> _FileIdentity:
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
try:
descriptor = os.open(path, flags)
except OSError as exc:
raise LidarGroundError("LiDAR evidence file is unavailable or unsafe") from exc
try:
value = os.fstat(descriptor)
current = os.lstat(path)
if (
not stat.S_ISREG(value.st_mode)
or stat.S_ISLNK(current.st_mode)
or (current.st_dev, current.st_ino) != (value.st_dev, value.st_ino)
):
raise LidarGroundError("LiDAR evidence file changed during no-follow open")
return (
value.st_dev,
value.st_ino,
value.st_size,
value.st_mtime_ns,
value.st_ctime_ns,
)
finally:
os.close(descriptor)
def _private_directory(path: Path) -> Path:
path.mkdir(mode=0o700, parents=True, exist_ok=True)
metadata = path.lstat()
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
raise OSError("LiDAR validation cache root is unsafe")
return path
src/k1link/web/session_api.py
-12
View File
@@ -353,7 +353,6 @@ def build_session_router(
**(
{
"preparation": _catalog_preparation_document(
store,
recording_preparation_manager,
item.session_id,
item.replayable,
@@ -1530,7 +1529,6 @@ def _require_matching_recording_generation(
def _catalog_preparation_document(
store: SessionStore,
manager: SessionRecordingPreparationManager | None,
session_id: str,
replayable: bool,
@@ -1538,16 +1536,6 @@ def _catalog_preparation_document(
if manager is None or not replayable:
return None
snapshot = manager.status(session_id)
if snapshot is None:
try:
snapshot = manager.restore_published(store.prepare_replay(session_id))
except (
SessionNotFoundError,
SessionNotReplayableError,
SessionIntegrityError,
ValueError,
):
return None
if snapshot is None:
return None
document: dict[str, Any] = {