Spaces:
Running
Running
File size: 33,993 Bytes
e16348e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 | """Random-access byte source for chesstb tables served over plain HTTP,
via ``Range`` requests, instead of a local filesystem.
This module has no chesstb-specific logic of its own -- it's a generic
HTTP byte-range client -- and lives in remote/ as shared infrastructure
for remote (URL) ``TABLEBASE_PATH`` support, used by both remote backends
tablebase_router can select via ``config.REMOTE_MODE`` (see each file's module
docstring for its own design):
* remote/remote_direct.py ("direct", the default) resolves tables to
:class:`RemoteFile` handles; tablebase_router reads them through
:class:`RemoteFileView` below so only the byte ranges a probe actually
touches are ever fetched.
* remote/remote_fallback.py ("download") downloads each table file in
full on first touch and disk-caches it, using only
:class:`RemoteHTTPClient` and :class:`RemoteSourceError` from this
module.
Design
------
* :class:`RemoteHTTPClient` -- a thin ``requests.Session`` wrapper that
knows how to ask "does `<base_url>/<path>` exist, and how big is it?"
and "give me bytes [offset, offset+length) of `<base_url>/<path>`", and
keeps a running count of every HTTP request it issues and every
response byte it receives, available via :meth:`RemoteHTTPClient.stats`.
Its session mounts an ``HTTPAdapter`` sized to the caller's expected
concurrency (``pool_maxsize``, normally set from ``PROBE_THREADS``) so
concurrent requests reuse warm connections instead of opening fresh
ones once the default pool is exhausted, and its retry loop backs off
between attempts -- honoring a ``Retry-After`` header when the server
sends one -- rather than retrying instantly into a struggling or
rate-limiting CDN edge.
* :class:`_PageCache` -- a sharded LRU cache, shared by every table opened
against one remote base URL, of fixed-size raw page fetches. A page's
key is hashed to one of a fixed number of independent shards -- each
with its own lock, LRU ledger, per-key fetch lock, and slice of the
overall byte budget -- so concurrent reads of pages that land in
different shards never contend on a single mutex. This is what makes
the hot path fast: a table's index structures (consulted on every
single probe) do many small, scattered reads clustered in a fairly
small region of each table file, so after the first probe touches a
region its page is cached and every later probe against the same table
is served with no network round trip at all. Each shard also tracks its
own hit/miss counts, summed by :attr:`_PageCache.hits` /
:attr:`_PageCache.misses` for the whole cache.
* :class:`RemoteFile` -- one opened remote table (fixed client + path +
size), with a ``read(offset, length)`` that satisfies a request from
the shared page cache, fetching only whichever pages aren't already
resident.
* :class:`RemoteFileView` -- a lightweight ``(RemoteFile, base, length)``
view supporting the same slicing/indexing a local
``memoryview(mmap(...))`` buffer would. Slicing a view is free -- it
never touches the network; it narrows the (base, length) window.
Only an explicit materialisation (``bytes(view)``, iterating it, or
indexing a single byte) triggers a fetch, and only for the span
actually needed. This is what tablebase_router's file classes read through
in "direct" mode.
"""
from __future__ import annotations
import collections
import threading
import time
import weakref
from typing import Dict, List, Optional, Tuple
from urllib.parse import parse_qs, urlsplit
try:
import requests # type: ignore
except ImportError: # pragma: no cover - exercised only without `requests`
requests = None # type: ignore
__all__ = [
"RemoteSourceError",
"RemoteHTTPClient",
"RemoteFile",
"RemoteFileView",
"looks_like_remote",
"DEFAULT_PAGE_SIZE",
"DEFAULT_PAGE_CACHE_BYTES",
"DEFAULT_PAGE_CACHE_SHARDS",
"DEFAULT_TIMEOUT",
"DEFAULT_MAX_RETRIES",
"DEFAULT_POOL_MAXSIZE",
]
#: Size of one page fetched/cached at a time. Large enough that a table's
#: whole header + index-vector region (dict, MonoUintVec/Min0UintVec
#: offsets -- everything but the bulk compressed data) typically fits in
#: one or two pages, small enough that a single cold random probe doesn't
#: pull down an unreasonable amount of unrelated data.
DEFAULT_PAGE_SIZE = 256 * 1024 # 256 KiB
#: Soft budget (bytes) for the shared LRU page cache, mirroring
#: chesstb.py's DEFAULT_BLOCK_CACHE_BYTES for the decoded-block cache one
#: layer up. Configurable via Tablebase/open_tablebase kwargs.
DEFAULT_PAGE_CACHE_BYTES = 128 * 1024 * 1024 # 128 MiB
#: Number of independent shards the page cache splits its keyspace
#: across. Each shard gets its own lock and an equal slice of
#: DEFAULT_PAGE_CACHE_BYTES, so concurrent probe threads reading
#: different pages -- the common case once a session has more than a
#: handful of materials open -- don't serialize behind one mutex.
DEFAULT_PAGE_CACHE_SHARDS = 16
DEFAULT_TIMEOUT = 20.0
DEFAULT_MAX_RETRIES = 3
#: Default HTTP connection pool size for a client's underlying
#: ``requests.Session``, used when the caller (normally app.py, sizing
#: this against ``PROBE_THREADS``) does not specify one explicitly.
#: Each host gets its own pool; resolver and CDN connections stay reusable.
DEFAULT_POOL_MAXSIZE = 20
#: Cap (seconds) on the exponential backoff between retried requests --
#: see :func:`_retry_delay`.
_MAX_RETRY_DELAY = 4.0
#: Status codes worth backing off before retrying: 429 (rate limited) and
#: the transient gateway/server errors a CDN edge or origin can return
#: under load. 404 and other 4xx codes are not included -- those mean the
#: request itself won't succeed on retry, so there's nothing to wait for.
_RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
def looks_like_remote(directory: str) -> bool:
"""True if `directory` (the value the app would otherwise treat as a
local ``TABLEBASE_PATH``) is actually an ``http(s)://`` base URL --
e.g. a Hugging Face bucket such as
``https://huggingface.co/buckets/<owner>/<name>/resolve``."""
return bool(directory) and directory.startswith(("http://", "https://"))
def _retry_delay(attempt: int, resp: Optional["requests.Response"]) -> float:
"""Seconds to wait before retry number `attempt` (0-based).
Honors a numeric ``Retry-After`` header on `resp` if present (a
server explicitly asking for a pause takes priority over guessing);
otherwise a capped exponential backoff, so a burst of concurrent
requests hitting a struggling or rate-limiting CDN edge spread out
instead of retrying in lockstep.
"""
if resp is not None:
retry_after = resp.headers.get("Retry-After")
if retry_after is not None and retry_after.isdigit():
return float(retry_after)
return min(0.25 * (2 ** attempt), _MAX_RETRY_DELAY)
class RemoteSourceError(IOError):
"""Raised when a remote tablebase byte range can't be fetched."""
class _PageCacheShard:
"""One shard of :class:`_PageCache`: an independent LRU cache of raw,
still-compressed page-sized byte ranges, with its own hit/miss counts.
Structurally mirrors :class:`chesstb._BlockCache`: an ``OrderedDict``
used as an LRU ledger, evicted down to a soft byte budget, plus
per-key locks so concurrent probe threads landing on the same
not-yet-fetched page fetch it once rather than once each.
"""
def __init__(self, max_bytes: int) -> None:
self.max_bytes = max_bytes
self.cur_bytes = 0
self.hits = 0
self.misses = 0
self._data: Dict[Tuple[str, int], bytes] = {}
self._lru: "collections.OrderedDict[Tuple[str, int], int]" = collections.OrderedDict()
self._lock = threading.Lock()
self._page_locks: weakref.WeakValueDictionary = weakref.WeakValueDictionary()
self._meta_lock = threading.Lock()
def lock_for(self, key: Tuple[str, int]) -> threading.Lock:
# The caller holds a strong reference while waiting/reading. A
# weak registry retains the same lock for every concurrent caller
# but releases it once nobody uses it, independently of eviction.
with self._meta_lock:
lk = self._page_locks.get(key)
if lk is None:
lk = threading.Lock()
self._page_locks[key] = lk
return lk
def get(self, key: Tuple[str, int]) -> Optional[bytes]:
with self._lock:
data = self._data.get(key)
if data is not None:
self._lru.move_to_end(key)
self.hits += 1
else:
self.misses += 1
return data
def put(self, key: Tuple[str, int], data: bytes) -> None:
with self._lock:
old = self._lru.pop(key, None)
if old is not None:
self.cur_bytes -= old
self._data[key] = data
self._lru[key] = len(data)
self.cur_bytes += len(data)
while self.cur_bytes > self.max_bytes and len(self._lru) > 1:
ev_key, ev_size = self._lru.popitem(last=False)
self._data.pop(ev_key, None)
self.cur_bytes -= ev_size
# Lock lifetime follows active users, not cached bytes.
def clear(self) -> None:
with self._lock:
self._data.clear()
self._lru.clear()
self.cur_bytes = 0
self.hits = 0
self.misses = 0
# Do not reset lock identities: an in-flight fetch (and its
# waiters) must still serialize with a read arriving after clear.
# Weak entries disappear automatically when the last user exits.
class _PageCache:
"""Sharded LRU cache of raw, still-compressed page-sized byte ranges
fetched over HTTP, shared by every table opened against one remote
base URL. See the module docstring for why this matters.
A page's ``(path, page_index)`` key is hashed to one of
:data:`DEFAULT_PAGE_CACHE_SHARDS` independent :class:`_PageCacheShard`
instances, each owning its own lock, LRU ledger, per-key fetch locks,
and an equal slice of the overall byte budget -- so a given key always
lands on the same shard (its cache entry and its fetch lock are always
found together), and reads or writes to pages in different shards
never contend on a shared mutex. ``max_bytes``, ``cur_bytes``,
``page_count``, ``hits``, and ``misses`` each report totals summed
across every shard.
"""
def __init__(self, max_bytes: int) -> None:
self._shard_count = DEFAULT_PAGE_CACHE_SHARDS
per_shard = max(max_bytes // self._shard_count, 1)
self._shards: List[_PageCacheShard] = [
_PageCacheShard(per_shard) for _ in range(self._shard_count)
]
def _shard(self, key: Tuple[str, int]) -> _PageCacheShard:
return self._shards[hash(key) % self._shard_count]
def lock_for(self, key: Tuple[str, int]) -> threading.Lock:
return self._shard(key).lock_for(key)
def get(self, key: Tuple[str, int]) -> Optional[bytes]:
return self._shard(key).get(key)
def put(self, key: Tuple[str, int], data: bytes) -> None:
self._shard(key).put(key, data)
def clear(self) -> None:
for shard in self._shards:
shard.clear()
@property
def max_bytes(self) -> int:
return sum(shard.max_bytes for shard in self._shards)
@property
def cur_bytes(self) -> int:
return sum(shard.cur_bytes for shard in self._shards)
@property
def page_count(self) -> int:
return sum(len(shard._lru) for shard in self._shards)
@property
def hits(self) -> int:
return sum(shard.hits for shard in self._shards)
@property
def misses(self) -> int:
return sum(shard.misses for shard in self._shards)
class RemoteHTTPClient:
"""Ranged-GET / existence-check client for one remote base URL, with a
running count of every HTTP request it issues and every response byte
it receives -- see :meth:`stats`."""
def __init__(self, base_url: str, *, timeout: float = DEFAULT_TIMEOUT,
max_retries: int = DEFAULT_MAX_RETRIES,
pool_maxsize: int = DEFAULT_POOL_MAXSIZE) -> None:
if requests is None:
raise RuntimeError(
"Remote tablebase access requires the 'requests' package. "
"Install it with `pip install requests` (already listed in "
"requirements.txt)."
)
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.max_retries = max(1, max_retries)
self._session = requests.Session()
# Keep resolver and CDN host pools, each sized for expected concurrency
# rather than left at requests' default of 10, which PROBE_THREADS
# alone can exceed. max_retries=0 here: retrying is this class's
# own job (see get_range/get_first_page), not urllib3's.
adapter = requests.adapters.HTTPAdapter(
pool_connections=4, pool_maxsize=max(1, pool_maxsize), max_retries=0,
)
self._session.mount("https://", adapter)
self._session.mount("http://", adapter)
self._stats_lock = threading.Lock()
self._requests = 0
self._bytes_fetched = 0
self._resolve_lock = threading.Lock()
self._resolved: Dict[str, Tuple[str, float]] = {}
def _record(self, nbytes: int = 0) -> None:
"""Count one HTTP round trip -- successful or not -- and however
many response body bytes it actually carried."""
with self._stats_lock:
self._requests += 1
self._bytes_fetched += nbytes
def close(self) -> None:
"""Release the session's connection pool. Safe to call more than
once -- ``requests.Session.close()`` itself is idempotent."""
self._session.close()
with self._resolve_lock:
self._resolved.clear()
def stats(self) -> Dict[str, int]:
"""Total HTTP requests issued and response bytes received through
this client so far."""
with self._stats_lock:
return {"requests": self._requests, "bytes_fetched": self._bytes_fetched}
def url_for(self, path: str) -> str:
return f"{self.base_url}/{path.lstrip('/')}"
def _request(self, method: str, path: str, **kwargs):
"""Reuse signed redirects until shortly before expiry.
Cache only successful redirects with an explicit expiry. Failed
cached URLs are retried through the original resolver, so an expired
CDN response cannot become a cached missing-table result.
"""
origin = self.url_for(path)
with self._resolve_lock:
cached = self._resolved.get(origin)
if cached is not None and cached[1] <= time.time():
self._resolved.pop(origin, None)
cached = None
kwargs.setdefault("allow_redirects", True)
if cached is not None:
try:
response = self._session.request(method, cached[0], **kwargs)
except requests.RequestException:
response = None
if response is not None and 200 <= response.status_code < 300:
return response
self._record()
if response is not None:
response.close()
with self._resolve_lock:
if self._resolved.get(origin) == cached:
self._resolved.pop(origin, None)
response = self._session.request(method, origin, **kwargs)
if 200 <= response.status_code < 300 and response.url != origin:
try:
expiry = int(parse_qs(urlsplit(response.url).query)["Expires"][0]) - 60
except (KeyError, ValueError, IndexError):
pass
else:
if expiry > time.time():
with self._resolve_lock:
self._resolved[origin] = (response.url, expiry)
return response
def head_size(self, path: str) -> Optional[int]:
"""Return `path`'s size in bytes, or ``None`` if it doesn't exist.
Tries a plain HEAD first; falls back to a 1-byte ranged GET (whose
``Content-Range`` header also carries the total size) for servers
that answer HEAD inconsistently for range-servable objects. Each
attempt tries HEAD-then-ranged-GET as one unit, retried with the
same backoff policy as get_range()/get_first_page(), so a
transient failure during size discovery doesn't skip the retry
budget those two already get.
"""
url = self.url_for(path)
last_exc: Optional[Exception] = None
for attempt in range(self.max_retries):
try:
resp = self._request("HEAD", path, timeout=self.timeout)
except requests.RequestException:
resp = None # fall through to the ranged-GET probe below
if resp is not None:
self._record() # HEAD has no body -- nothing to add to bytes_fetched
if resp.status_code == 404:
return None
if resp.status_code < 400:
cl = resp.headers.get("Content-Length")
if cl is not None and cl.isdigit():
return int(cl)
try:
resp = self._request("GET", path, headers={"Range": "bytes=0-0"}, timeout=self.timeout)
except requests.RequestException as exc:
self._record()
last_exc = RemoteSourceError(f"Failed to reach {url}: {exc}")
last_exc.__cause__ = exc
if attempt + 1 < self.max_retries:
time.sleep(_retry_delay(attempt, None))
continue
self._record(len(resp.content))
if resp.status_code == 404:
return None
if resp.status_code == 206:
content_range = resp.headers.get("Content-Range", "")
total = content_range.rsplit("/", 1)[-1] if "/" in content_range else ""
if total.isdigit():
return int(total)
# Unknown ("*") or unparsable total -- this method is the
# size-of-last-resort probe (see get_first_page()'s copy
# of this same situation, which falls back to this
# method), so there's nowhere further to fall back to.
last_exc = RemoteSourceError(
f"{url} returned 206 with an unparsable Content-Range total: {content_range!r}"
)
elif resp.status_code == 200:
# Server ignored Range and returned the whole object;
# expected only for a tiny or placeholder file.
return len(resp.content)
else:
last_exc = RemoteSourceError(f"Unexpected status {resp.status_code} probing {url}")
if resp.status_code in _RETRYABLE_STATUS_CODES and attempt + 1 < self.max_retries:
time.sleep(_retry_delay(attempt, resp))
raise last_exc or RemoteSourceError(f"Failed to probe size for {url}")
def get_range(self, path: str, offset: int, length: int) -> bytes:
if length <= 0:
return b""
url = self.url_for(path)
end = offset + length - 1
last_exc: Optional[Exception] = None
for attempt in range(self.max_retries):
try:
resp = self._request(
"GET", path, headers={"Range": f"bytes={offset}-{end}"},
timeout=self.timeout, stream=True,
)
except requests.RequestException as exc:
self._record()
# Normalised to RemoteSourceError immediately (rather than
# storing the raw requests exception) so every caller up
# the stack -- CompositeTablebase._find()'s fallback walk,
# the Flask routes' 502 mapping -- only ever has to catch
# one exception family for a transport failure, the same
# one already used for a bad HTTP status below.
last_exc = RemoteSourceError(f"Failed to reach {url}: {exc}")
last_exc.__cause__ = exc
if attempt + 1 < self.max_retries:
time.sleep(_retry_delay(attempt, None))
continue
if resp.status_code == 206:
self._record(len(resp.content))
return resp.content
if resp.status_code == 200:
# Range not honoured by this server/CDN -- stream=True
# above means the body isn't downloaded yet at this point,
# so a declared (or entirely undeclared -- chunked
# transfer-encoding, or a server that omits
# Content-Length) size can still be checked before
# `.content` below pulls it fully into memory. A
# compliant 206 always carries Content-Range and a body
# sized to the request; a 200 that comes back this large
# (or of unknown size) means the whole object came back,
# which for a multi-GB tablebase file is not something to
# buffer to slice a few hundred KiB out of it.
cl = resp.headers.get("Content-Length")
declared_size = int(cl) if cl is not None and cl.isdigit() else None
if declared_size is None or declared_size > length:
resp.close()
raise RemoteSourceError(
f"{url} ignored the Range request and returned status 200 "
f"with {'an unknown-size' if declared_size is None else declared_size} "
f"body (requested {length} bytes) -- refusing to buffer "
f"the full object in memory."
)
self._record(len(resp.content))
return resp.content[offset:offset + length]
if resp.status_code == 404:
resp.close()
raise RemoteSourceError(f"{url} not found")
self._record(len(resp.content) if resp.content is not None else 0)
last_exc = RemoteSourceError(
f"Unexpected status {resp.status_code} fetching {url} bytes={offset}-{end}"
)
if resp.status_code in _RETRYABLE_STATUS_CODES and attempt + 1 < self.max_retries:
time.sleep(_retry_delay(attempt, resp))
raise last_exc or RemoteSourceError(f"Failed to fetch {url} bytes={offset}-{end}")
def get_first_page(self, path: str, page_size: int) -> Tuple[Optional[int], bytes]:
"""Fetch `path`'s first `page_size` bytes, returning
``(total_size, data)`` -- ``total_size`` is ``None`` if `path`
doesn't exist, in which case ``data`` is empty.
This combines existence, size, and header retrieval in one round
trip: a freshly touched table needs all three values, and the
initial page is already required for header parsing. The returned
data is reused for the first page of the remote file. `data` is capped to `page_size` bytes
even if the server ignores the Range header and returns a whole
object no bigger than `page_size` -- a genuinely small file, the
one case worth still buffering. A larger (or size-unknown, e.g.
chunked) object coming back whole instead raises
:class:`RemoteSourceError` rather than buffering it in full; see
the ``status_code == 200`` branch below.
"""
if page_size <= 0:
return None, b""
url = self.url_for(path)
last_exc: Optional[Exception] = None
for attempt in range(self.max_retries):
try:
resp = self._request(
"GET", path, headers={"Range": f"bytes=0-{page_size - 1}"},
timeout=self.timeout, stream=True,
)
except requests.RequestException as exc:
self._record()
# See get_range()'s copy of this handler for why this is
# normalised to RemoteSourceError immediately.
last_exc = RemoteSourceError(f"Failed to reach {url}: {exc}")
last_exc.__cause__ = exc
if attempt + 1 < self.max_retries:
time.sleep(_retry_delay(attempt, None))
continue
if resp.status_code == 404:
self._record()
return None, b""
if resp.status_code == 206:
self._record(len(resp.content))
content_range = resp.headers.get("Content-Range", "")
total_str = content_range.rsplit("/", 1)[-1] if "/" in content_range else ""
if total_str.isdigit():
return int(total_str), resp.content[:page_size]
# Content-Range's total is "*" (RFC 9110 permits this when
# the sender doesn't know the complete representation
# length yet) or otherwise unparsable -- either way this
# means "exists, size unknown from this response", not
# "doesn't exist" (404 already returned above for that).
# Caching that distinction matters: the caller
# (_remote_size() in remote_direct.py) stores whatever
# this returns and, if it were None, would permanently
# treat an existing table as missing. head_size() is a
# separate probe that a server unable to state the range
# total inline may still be able to answer.
total = self.head_size(path)
if total is None:
raise RemoteSourceError(
f"{url} returned 206 with Content-Range {content_range!r} and "
f"a follow-up size probe found nothing at that path"
)
return total, resp.content[:page_size]
if resp.status_code == 200:
# Range not honoured -- could be a genuinely tiny file
# (smaller than page_size, the legitimate case the
# docstring above describes) or a large-object-ignoring-
# Range CDN returning the whole thing from byte 0. As in
# get_range(), stream=True above means Content-Length can
# be checked before `.content` reads the body, so only the
# first, legitimate case actually gets buffered here.
cl = resp.headers.get("Content-Length")
declared_size = int(cl) if cl is not None and cl.isdigit() else None
if declared_size is None or declared_size > page_size:
resp.close()
raise RemoteSourceError(
f"{url} ignored the Range request and returned status 200 "
f"with {'an unknown-size' if declared_size is None else declared_size} "
f"body (page_size is {page_size} bytes) -- refusing to buffer "
f"the full object in memory."
)
self._record(len(resp.content))
return len(resp.content), resp.content[:page_size]
self._record(len(resp.content) if resp.content is not None else 0)
last_exc = RemoteSourceError(f"Unexpected status {resp.status_code} probing {url}")
if resp.status_code in _RETRYABLE_STATUS_CODES and attempt + 1 < self.max_retries:
time.sleep(_retry_delay(attempt, resp))
raise last_exc or RemoteSourceError(f"Failed to fetch {url} bytes=0-{page_size - 1}")
class RemoteFile:
"""One opened remote table file: fixed (client, path, size), backed
by a page cache shared across every table opened against the same
remote base URL (see :class:`chesstb.Tablebase`)."""
def __init__(self, client: RemoteHTTPClient, path: str, size: int,
page_cache: _PageCache, page_size: int = DEFAULT_PAGE_SIZE) -> None:
self.client = client
self.path = path
self.size = size
self._page_cache = page_cache
self.page_size = page_size
def __len__(self) -> int:
return self.size
def __str__(self) -> str:
return self.client.url_for(self.path)
def _get_page(self, page_index: int) -> bytes:
key = (self.path, page_index)
data = self._page_cache.get(key)
if data is not None:
return data
with self._page_cache.lock_for(key):
data = self._page_cache.get(key)
if data is not None:
return data
start = page_index * self.page_size
length = min(self.page_size, self.size - start)
data = self.client.get_range(self.path, start, length)
self._page_cache.put(key, data)
return data
def read(self, offset: int, length: int) -> bytes:
"""Return up to `length` bytes starting at `offset`, clamped to
the file's actual size (mirrors slicing past the end of a
``memoryview``/``mmap`` rather than raising)."""
if length <= 0 or offset >= self.size:
return b""
length = min(length, self.size - offset)
first_page = offset // self.page_size
last_page = (offset + length - 1) // self.page_size
if first_page == last_page:
page = self._get_page(first_page)
start_in_page = offset - first_page * self.page_size
return page[start_in_page:start_in_page + length]
parts = []
remaining = length
cur_offset = offset
for page_index in range(first_page, last_page + 1):
page = self._get_page(page_index)
page_start = page_index * self.page_size
# The file's own last page is legitimately shorter than
# page_size whenever self.size isn't a multiple of it --
# that's normal, not truncation. expected_page_len is what
# this page should be regardless, so it's still page_size for
# every non-final page.
expected_page_len = min(self.page_size, self.size - page_start)
start_in_page = cur_offset - page_start
# max(0, ...): a page shorter than its own expected_page_len
# (a truncated remote response) would otherwise make
# len(page) - start_in_page negative -- Python accepts a
# negative slice bound rather than raising, so without this
# guard `take` would go negative and cur_offset/remaining
# would move backward/grow instead of making forward
# progress.
take = max(0, min(len(page) - start_in_page, remaining))
if take > 0:
parts.append(page[start_in_page:start_in_page + take])
cur_offset += take
remaining -= take
if len(page) < expected_page_len:
# This page came back shorter than its own expected
# length -- not only take == 0 above, which catches
# the case where this page had nothing left to give at
# cur_offset's position. A short (but nonempty) page
# still needs to stop the read here: every later
# iteration's start_in_page assumes cur_offset has
# advanced exactly page_size per prior page, which a
# short page breaks -- continuing would compute a
# negative (and silently wrong-but-not-erroring, per the
# same slicing behavior above) start_in_page against the
# next page. Ending the read here instead mirrors this
# method's own documented past-EOF slicing behavior.
break
return b"".join(parts)
def close(self) -> None:
# Fetched pages live in the Tablebase-shared page cache, not
# here -- nothing owned by this object needs releasing.
pass
class RemoteFileView:
"""A lazy, ``memoryview``-slicing-compatible view over a
:class:`RemoteFile`. See the module docstring for why slicing this is
free and only explicit materialisation fetches bytes.
"""
__slots__ = ("_file", "_base", "_len")
def __init__(self, remote_file: RemoteFile, base: int = 0,
length: Optional[int] = None) -> None:
self._file = remote_file
self._base = base
self._len = length if length is not None else (len(remote_file) - base)
def __len__(self) -> int:
return self._len
def __getitem__(self, item):
if isinstance(item, slice):
start, stop, step = item.indices(self._len)
if step != 1:
raise ValueError("RemoteFileView does not support strided slicing")
return RemoteFileView(self._file, self._base + start, max(0, stop - start))
idx = item + self._len if item < 0 else item
if not (0 <= idx < self._len):
raise IndexError("RemoteFileView index out of range")
chunk = self._file.read(self._base + idx, 1)
if not chunk:
raise IndexError("RemoteFileView index out of range (past end of remote file)")
return chunk[0]
def get_bytes(self) -> bytes:
"""Materialise (fetch, via the page cache) this whole view's
bytes. Slice first (``view[a:b]``) to materialise a sub-range."""
return self._file.read(self._base, self._len)
def __bytes__(self) -> bytes:
return self.get_bytes()
|