Asyncfs

asyncfs

Async byte access for omfiles: HTTP pages with memory and disk caches, or a mirror.

omfiles reads an archive file through an fsspec-like object exposing _cat_file/_size. Its reads are tiny and many, so :class:PagedAsyncFS serves them from aligned pages (one download per page, shared by concurrent readers), kept in memory and, for files that no longer change, on disk (:class:DiskPages). :class:LocalAsyncFS reads a local mirror directly.

DiskPages dataclass

Where :class:PagedAsyncFS keeps the pages of immutable files.

Source code in packages/barograph-met/src/baro/met/asyncfs.py

@dataclass(frozen=True)
class DiskPages:
    """Where :class:`PagedAsyncFS` keeps the pages of immutable files."""

    cache: RangeCache
    namespace: str
    cacheable: Callable[[str], bool]

    @classmethod
    def at(cls, cache: RangeCache, location: str, cacheable: Callable[[str], bool]) -> DiskPages:
        """Pages of the files under ``location`` (a URL) whose path passes ``cacheable``."""
        return cls(cache, namespace_for(location), cacheable)

at(cache, location, cacheable) classmethod

Pages of the files under location (a URL) whose path passes cacheable.

Source code in packages/barograph-met/src/baro/met/asyncfs.py

@classmethod
def at(cls, cache: RangeCache, location: str, cacheable: Callable[[str], bool]) -> DiskPages:
    """Pages of the files under ``location`` (a URL) whose path passes ``cacheable``."""
    return cls(cache, namespace_for(location), cacheable)

LocalAsyncFS

The same interface over a local mirror directory.

Source code in packages/barograph-met/src/baro/met/asyncfs.py

class LocalAsyncFS:
    """The same interface over a local mirror directory."""

    def __init__(self, root: str | Path, stats: TransferStats) -> None:
        self.root = Path(root)
        self.stats = stats

    async def _cat_file(
        self, path: str, start: int | None = None, end: int | None = None, **_: Any
    ) -> bytes:
        file = self.root / path
        if not file.is_file():
            raise FileNotFoundError(path)
        with file.open("rb") as fh:
            fh.seek(start or 0)
            data = fh.read(None if end is None else end - (start or 0))
        self.stats.add(len(data), 0.0)
        return data

    async def _size(self, path: str, **_: Any) -> int:
        file = self.root / path
        if not file.is_file():
            raise FileNotFoundError(path)
        return file.stat().st_size

    async def aclose(self) -> None:
        """Nothing to release."""

aclose() async

Nothing to release.

Source code in packages/barograph-met/src/baro/met/asyncfs.py

async def aclose(self) -> None:
    """Nothing to release."""

PagedAsyncFS

Async fsspec-like object for omfiles (_cat_file/_size), with a page cache.

omfiles issues one range request per internal data chunk, and archive chunks are tiny (1 latitude row x 14 longitudes, a few hundred bytes). Requests are therefore served from aligned pages of page_size bytes (a page holds several whole latitude rows); concurrent requests for the same page share one download. This trades a few extra bytes for ~50x fewer round trips, which dominate the cost.

Files for which immutable(path) is false may be rewritten upstream: their version (size and ETag) is re-checked after :data:REVALIDATE_S seconds and pages are keyed by it, so a read never mixes two versions.

Source code in packages/barograph-met/src/baro/met/asyncfs.py

class PagedAsyncFS:
    """Async fsspec-like object for ``omfiles`` (``_cat_file``/``_size``), with a page cache.

    ``omfiles`` issues one range request per internal data chunk, and archive
    chunks are tiny (1 latitude row x 14 longitudes, a few hundred bytes).
    Requests are therefore served from aligned pages of ``page_size`` bytes
    (a page holds several whole latitude rows); concurrent requests for the
    same page share one download. This trades a few extra bytes for ~50x fewer
    round trips, which dominate the cost.

    Files for which ``immutable(path)`` is false may be rewritten upstream:
    their version (size and ETag) is re-checked after :data:`REVALIDATE_S`
    seconds and pages are keyed by it, so a read never mixes two versions.
    """

    def __init__(
        self,
        base_url: str,
        stats: TransferStats,
        max_connections: int,
        page_size: int = 512 << 10,
        cache_mb: float = 256.0,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.stats = stats
        self.page_size = page_size
        self.client = httpx.AsyncClient(
            timeout=120.0,
            follow_redirects=True,
            limits=httpx.Limits(max_connections=max_connections),
            headers={"User-Agent": USER_AGENT},
        )
        self._pages: OrderedDict[tuple[str, str, int], bytes] = OrderedDict()
        self._inflight: dict[tuple[str, str, int], asyncio.Future[bytes]] = {}
        self._versions: dict[str, tuple[int, str, float]] = {}  # size, tag, checked at
        self.immutable: Callable[[str], bool] = lambda _path: False
        self._bytes = 0
        self._limit = int(cache_mb * 1e6)
        self.disk: DiskPages | None = None  # immutable pages also kept on disk

    async def _request(self, method: str, path: str, headers: dict[str, str]) -> httpx.Response:
        url = f"{self.base_url}/{path}"
        for attempt in range(_ATTEMPTS):
            try:
                t0 = time.perf_counter()
                async with async_slot(url):
                    resp = await self.client.request(method, url, headers=headers)
            except httpx.TransportError:
                if attempt == _ATTEMPTS - 1:
                    raise
                await asyncio.sleep(0.5 * 2**attempt)
                continue
            if resp.status_code in _RETRY_STATUS and attempt < _ATTEMPTS - 1:
                self.stats.throttle()
                _log.warning(
                    "HTTP %s from %s (attempt %d), backing off", resp.status_code, url, attempt + 1
                )
                retry_after = resp.headers.get("retry-after", "")
                delay = float(retry_after) if retry_after.isdigit() else 0.5 * 2**attempt
                await asyncio.sleep(min(delay, 60.0))
                continue
            if resp.status_code == HTTPStatus.NOT_FOUND:
                raise FileNotFoundError(path)
            resp.raise_for_status()
            self.stats.add(len(resp.content), time.perf_counter() - t0)
            return resp
        raise RuntimeError("unreachable")  # pragma: no cover

    async def _page(self, path: str, tag: str, index: int, size: int) -> bytes:
        key = (path, tag, index)
        if key in self._pages:
            self._pages.move_to_end(key)
            return self._pages[key]
        if key in self._inflight:
            return await self._inflight[key]
        future: asyncio.Future[bytes] = asyncio.get_running_loop().create_future()
        self._inflight[key] = future
        try:
            data = await self._load(path, index, size)
            self._remember(key, data)
            future.set_result(data)
            return data
        except BaseException as exc:
            future.set_exception(exc)
            raise
        finally:
            del self._inflight[key]

    async def _load(self, path: str, index: int, size: int) -> bytes:
        """Page ``index`` of ``path``: from disk if kept there, else downloaded (and kept)."""
        start = index * self.page_size
        stop = min(start + self.page_size, size)
        disk = self.disk if self.disk is not None and self.disk.cacheable(path) else None
        data = None if disk is None else disk.cache.lookup(disk.namespace, path, start, stop)
        if data is None:
            resp = await self._request("GET", path, {"Range": f"bytes={start}-{stop - 1}"})
            data = resp.content
            if resp.status_code != HTTPStatus.PARTIAL_CONTENT or len(data) != stop - start:
                raise RuntimeError(
                    f"{path}: asked for bytes {start}-{stop - 1}, got HTTP "
                    f"{resp.status_code} with {len(data)} bytes (range ignored or file changed)"
                )
            if disk is not None:
                disk.cache.put(disk.namespace, path, start, data)
        return data

    def _remember(self, key: tuple[str, str, int], data: bytes) -> None:
        """Keep a page in memory, evicting the least recently used beyond the limit."""
        self._pages[key] = data
        self._bytes += len(data)
        while self._bytes > self._limit and len(self._pages) > 1:
            _, old = self._pages.popitem(last=False)
            self._bytes -= len(old)

    async def _cat_file(
        self, path: str, start: int | None = None, end: int | None = None, **_: Any
    ) -> bytes:
        size, tag = await self._version(path)
        start = 0 if start is None else start
        end = size if end is None else min(end, size)
        first, last = start // self.page_size, max(end - 1, start) // self.page_size
        pages = await asyncio.gather(
            *(self._page(path, tag, i, size) for i in range(first, last + 1))
        )
        blob = b"".join(pages)
        offset = first * self.page_size
        return blob[start - offset : end - offset]

    async def _size(self, path: str, **_: Any) -> int:
        return (await self._version(path))[0]

    async def _version(self, path: str) -> tuple[int, str]:
        """Size and version tag of ``path``; re-checked after a while unless immutable."""
        known = self._versions.get(path)
        if known is not None and (
            self.immutable(path) or time.monotonic() - known[2] < REVALIDATE_S
        ):
            return known[0], known[1]
        resp = await self._request("HEAD", path, {})
        size = int(resp.headers["content-length"])
        tag = resp.headers.get("etag") or resp.headers.get("last-modified") or str(size)
        self._versions[path] = (size, tag, time.monotonic())
        return size, tag

    async def aclose(self) -> None:
        """Close the HTTP connection pool."""
        await self.client.aclose()

aclose() async

Close the HTTP connection pool.

Source code in packages/barograph-met/src/baro/met/asyncfs.py

async def aclose(self) -> None:
    """Close the HTTP connection pool."""
    await self.client.aclose()