kqjy/MyFSIO
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 33 Wiki Activity

Debug object browser object count delay

Browse Source
This commit is contained in:
kqjy
2026-02-05 21:08:18 +08:00
parent e9a035827b
commit 2643a79121
1 changed files with 21 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
app/storage.py
View File

@@ -322,19 +322,23 @@ class ObjectStorage:
"""Incrementally update cached bucket statistics instead of invalidating. """Incrementally update cached bucket statistics instead of invalidating.
This avoids expensive full directory scans on every PUT/DELETE by This avoids expensive full directory scans on every PUT/DELETE by
adjusting the cached values directly. adjusting the cached values directly. Also signals cross-process cache
invalidation by updating the file mtime.
""" """
cache_path = self._system_bucket_root(bucket_id) / "stats.json" cache_path = self._system_bucket_root(bucket_id) / "stats.json"
try: try:
cache_path.parent.mkdir(parents=True, exist_ok=True)
if cache_path.exists(): if cache_path.exists():
data = json.loads(cache_path.read_text(encoding="utf-8")) data = json.loads(cache_path.read_text(encoding="utf-8"))
data["objects"] = max(0, data.get("objects", 0) + objects_delta) else:
data["bytes"] = max(0, data.get("bytes", 0) + bytes_delta) data = {"objects": 0, "bytes": 0, "version_count": 0, "version_bytes": 0, "total_objects": 0, "total_bytes": 0}
data["version_count"] = max(0, data.get("version_count", 0) + version_count_delta) data["objects"] = max(0, data.get("objects", 0) + objects_delta)
data["version_bytes"] = max(0, data.get("version_bytes", 0) + version_bytes_delta) data["bytes"] = max(0, data.get("bytes", 0) + bytes_delta)
data["total_objects"] = max(0, data.get("total_objects", 0) + objects_delta + version_count_delta) data["version_count"] = max(0, data.get("version_count", 0) + version_count_delta)
data["total_bytes"] = max(0, data.get("total_bytes", 0) + bytes_delta + version_bytes_delta) data["version_bytes"] = max(0, data.get("version_bytes", 0) + version_bytes_delta)
cache_path.write_text(json.dumps(data), encoding="utf-8") data["total_objects"] = max(0, data.get("total_objects", 0) + objects_delta + version_count_delta)
data["total_bytes"] = max(0, data.get("total_bytes", 0) + bytes_delta + version_bytes_delta)
cache_path.write_text(json.dumps(data), encoding="utf-8")
except (OSError, json.JSONDecodeError): except (OSError, json.JSONDecodeError):
pass pass
@@ -1658,34 +1662,27 @@ class ObjectStorage:
"""Invalidate the object cache and etag index for a bucket. """Invalidate the object cache and etag index for a bucket.
Increments version counter to signal stale reads. Increments version counter to signal stale reads.
Also touches marker file for cross-process invalidation. Cross-process invalidation is handled by checking stats.json mtime.
""" """
with self._cache_lock: with self._cache_lock:
self._object_cache.pop(bucket_id, None) self._object_cache.pop(bucket_id, None)
self._cache_version[bucket_id] = self._cache_version.get(bucket_id, 0) + 1 self._cache_version[bucket_id] = self._cache_version.get(bucket_id, 0) + 1
self._touch_cache_marker(bucket_id)
etag_index_path = self._system_bucket_root(bucket_id) / "etag_index.json" etag_index_path = self._system_bucket_root(bucket_id) / "etag_index.json"
try: try:
etag_index_path.unlink(missing_ok=True) etag_index_path.unlink(missing_ok=True)
except OSError: except OSError:
pass pass
def _touch_cache_marker(self, bucket_id: str) -> None:
"""Touch the cache marker file to signal other processes that cache is stale."""
marker_path = self._system_bucket_root(bucket_id) / ".cache_marker"
try:
marker_path.parent.mkdir(parents=True, exist_ok=True)
marker_path.write_text(str(time.time()), encoding="utf-8")
except OSError:
pass
def _get_cache_marker_mtime(self, bucket_id: str) -> float: def _get_cache_marker_mtime(self, bucket_id: str) -> float:
"""Get the mtime of the cache marker file, or 0 if it doesn't exist.""" """Get the mtime of stats.json for cross-process cache invalidation.
marker_path = self._system_bucket_root(bucket_id) / ".cache_marker"
Uses stats.json because it's already updated on every object change
via _update_bucket_stats_cache.
"""
stats_path = self._system_bucket_root(bucket_id) / "stats.json"
try: try:
return marker_path.stat().st_mtime return stats_path.stat().st_mtime
except OSError: except OSError:
return 0 return 0
@@ -1693,7 +1690,7 @@ class ObjectStorage:
"""Update a single entry in the object cache instead of invalidating the whole cache. """Update a single entry in the object cache instead of invalidating the whole cache.
This is a performance optimization - lazy update instead of full invalidation. This is a performance optimization - lazy update instead of full invalidation.
Also touches a marker file to signal cache invalidation to other processes. Cross-process invalidation is handled by checking stats.json mtime.
""" """
with self._cache_lock: with self._cache_lock:
cached = self._object_cache.get(bucket_id) cached = self._object_cache.get(bucket_id)
@@ -1705,7 +1702,6 @@ class ObjectStorage:
objects[key] = meta objects[key] = meta
self._cache_version[bucket_id] = self._cache_version.get(bucket_id, 0) + 1 self._cache_version[bucket_id] = self._cache_version.get(bucket_id, 0) + 1
self._sorted_key_cache.pop(bucket_id, None) self._sorted_key_cache.pop(bucket_id, None)
self._touch_cache_marker(bucket_id)
def warm_cache(self, bucket_names: Optional[List[str]] = None) -> None: def warm_cache(self, bucket_names: Optional[List[str]] = None) -> None:
"""Pre-warm the object cache for specified buckets or all buckets. """Pre-warm the object cache for specified buckets or all buckets.