From e9a4ab70888aa4ca70cfeb3ecefac5957e1e033c Mon Sep 17 00:00:00 2001 From: Grigori Fursin Date: Fri, 9 Oct 2026 10:41:07 +0200 Subject: [PATCH 1/4] docs: the API pages regenerated by the 0.34.0 docs build (PathLock and its constants, the output decoding helpers) Co-Authored-By: Claude Fable 5.1 Signed-off-by: Grigori Fursin --- docs/en/api/cmeta_utils_files.rst | 24 ++++++++++++++++++++++++ docs/en/api/cmeta_utils_sys.rst | 4 ++++ 2 files changed, 28 insertions(+) diff --git a/docs/en/api/cmeta_utils_files.rst b/docs/en/api/cmeta_utils_files.rst index 38844d3..6169dd2 100644 --- a/docs/en/api/cmeta_utils_files.rst +++ b/docs/en/api/cmeta_utils_files.rst @@ -7,6 +7,14 @@ cMeta author and developer: (C) 2025-2026 Grigori Fursin See the cMeta COPYRIGHT and LICENSE files in the project root for details. +Classes +------- + +.. autoclass:: cmeta.utils.files.PathLock + :members: + :undoc-members: + :show-inheritance: + Functions --------- @@ -91,6 +99,22 @@ Constants .. autodata:: cmeta.utils.files.ERROR_CODE_FILE_NOT_FOUND +.. autodata:: cmeta.utils.files.LOCK_IDENTITY_RETRIES + +.. autodata:: cmeta.utils.files.LOCK_POLL_MAX + +.. autodata:: cmeta.utils.files.LOCK_POLL_START + +.. autodata:: cmeta.utils.files.LOCK_SOFT_CHECK_SECONDS + +.. autodata:: cmeta.utils.files.LOCK_SOFT_DEAD_MIN_AGE_SECONDS + +.. autodata:: cmeta.utils.files.LOCK_SOFT_EMPTY_STALE_SECONDS + +.. autodata:: cmeta.utils.files.LOCK_SOFT_REMOVE_SECONDS + +.. autodata:: cmeta.utils.files.LOCK_SOFT_STALE_SECONDS + .. autodata:: cmeta.utils.files.LOCK_SUFFIX .. autodata:: cmeta.utils.files.RETRY_DELAY diff --git a/docs/en/api/cmeta_utils_sys.rst b/docs/en/api/cmeta_utils_sys.rst index f67bdf5..f6ea2fe 100644 --- a/docs/en/api/cmeta_utils_sys.rst +++ b/docs/en/api/cmeta_utils_sys.rst @@ -12,6 +12,8 @@ Functions .. autofunction:: cmeta.utils.sys.cmeta_install_info +.. autofunction:: cmeta.utils.sys.decode_output + .. autofunction:: cmeta.utils.sys.describe_cmeta_install .. autofunction:: cmeta.utils.sys.find_command_func @@ -45,3 +47,5 @@ Functions .. autofunction:: cmeta.utils.sys.run .. autofunction:: cmeta.utils.sys.run_command_with_timeout_tree_kill_on_windows + +.. autofunction:: cmeta.utils.sys.system_code_pages From 142ef77411579c120dc29d3de0ad04b49129d9fc Mon Sep 17 00:00:00 2001 From: Grigori Fursin Date: Fri, 9 Oct 2026 13:47:37 +0200 Subject: [PATCH 2/4] The index lock: writes made during a full reindex wait for it and land in the new index, two rebuilds never race for the swap (0.34.0.1) - /index.lock (the sidecar of the index folder, kept through the swap) is held by the full reindex for its whole rebuild and swap, by the incremental index of pull/plug/repo reindex for its duration and by every write of a record for a moment; readers never take it - a writer that finds it held waits: a notice on stderr after 3 s naming the reindexing process (which leaves a note in the lock file), up to CMETA_INDEX_LOCK_TIMEOUT seconds (600), then an error - the first run on a home takes the lock before deciding that the index is missing, so several first runs at once build one index; the home and its local repository folder are created with exist_ok - under the lock a reindex removes every index.tmp-* / index.old-* leftover of killed rebuilds - PathLock.acquire(note=...), PathLock.write_note(), PathLock.read_note() - tests/internal_repo_tests/test_integration_index_guard.py: writers of every kind during a reindex paused in another process, two reindexes at once, four first runs on a fresh home, the leftovers, the notice and the timeout, the note - docs/async-and-concurrency.md "The index lock", AGENTS.md and docs/using-cmeta.md env lists, CHANGELOG Co-Authored-By: Claude Fable 5.1 Signed-off-by: Grigori Fursin --- AGENTS.md | 3 +- CHANGELOG.md | 22 ++ cmeta/repos.py | 168 ++++++++++++- cmeta/utils/files.py | 46 ++++ cmeta/version.py | 2 +- docs/async-and-concurrency.md | 25 +- docs/using-cmeta.md | 3 +- .../test_integration_index_guard.py | 228 ++++++++++++++++++ 8 files changed, 479 insertions(+), 18 deletions(-) create mode 100644 tests/internal_repo_tests/test_integration_index_guard.py diff --git a/AGENTS.md b/AGENTS.md index e0cc7c5..abff02d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -63,7 +63,8 @@ tabulate, tqdm, packaging, psutil. (File locks are the engine's own ### Useful environment variables `CMETA_HOME`, `CMETA_DEBUG=1`, `CMETA_LOG=DEBUG|INFO`, `CMETA_LOG_FILE=`, `CMETA_FAIL_ON_ERROR=yes`, `CMETA_VERBOSE=yes`, `CMETA_INTERNAL_REPO_PATH`, -`CMETA_PIP_INSTALL_ARGS`, and the provenance stamped on new artifacts: `CMETA_AUTHORS`, `CMETA_COPYRIGHT`, +`CMETA_PIP_INSTALL_ARGS`, `CMETA_INDEX_LOCK_TIMEOUT` (seconds a write to the index waits for a running +reindex, 600), and the provenance stamped on new artifacts: `CMETA_AUTHORS`, `CMETA_COPYRIGHT`, `CMETA_GENERATOR` (JSON: how the artifact was made; with the `artifact_defaults` of a repository's `_cmr.yaml` - docs/using-cmeta.md §7.6). diff --git a/CHANGELOG.md b/CHANGELOG.md index 200682f..6b49c59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,28 @@ All notable changes to cMeta are documented here, newest first. +## 0.34.0.1 (in development) +- **The index lock: a full reindex no longer loses the writes made while it runs, and two rebuilds no + longer race for the swap.** A full `cx --reindex` builds the new index in a temporary folder and swaps + it in when it is complete; a record written to the live index meanwhile by another process (a + create, update, delete or reindex of an artifact) was lost in the swap, and a deletion came back as a + record whose folder was gone. The lock file `/index.lock` - the sidecar of the index folder, + which the swap leaves in place - is now held by the full reindex for its whole rebuild and swap, by + the index of pulled or plugged repositories for its duration, and by every write of a record for a + moment: a writer that arrives during a reindex waits for it (a notice on stderr after 3 s naming the + reindexing process, which leaves a note in the lock file; `CMETA_INDEX_LOCK_TIMEOUT`, 600 s by + default, bounds the wait, then the write fails with an error) and then writes into the new index; two + reindexes run one after the other; several processes starting on a fresh home at once build one + index, the others wait and find it; a process that starts during the swap waits instead of + rebuilding. Under the same lock a reindex removes the `index.tmp-*` / `index.old-*` leftovers of + killed rebuilds (before, only its own). Readers never take the lock, so lookups cost what they did; a + write costs one more lock acquisition. `PathLock.acquire` takes a `note` and `PathLock.read_note` + reads it. Tests: `tests/internal_repo_tests/test_integration_index_guard.py` (writers of every kind + during a paused reindex, two reindexes at once, four first runs on a fresh home, the leftovers, the + notice and the timeout, the note). With it, the first run creates the home and its local repository + folder with `exist_ok`: two first runs on one home at once no longer fail each other with "cannot + create a file when that file already exists". + ## 0.34.0 - **The engine's own path lock replaces the `filelock` library: two processes can no longer hold one lock at once, and lock files are still removed after use.** The previous lock removed its file after diff --git a/cmeta/repos.py b/cmeta/repos.py index eb7ae58..3faa117 100644 --- a/cmeta/repos.py +++ b/cmeta/repos.py @@ -24,6 +24,14 @@ # How many stubs in a row a lookup follows before it gives up (a stub pointing to a stub) MAX_MIGRATED_HOPS = 8 +# The index lock (`/index.lock`): a full reindex holds it for its whole rebuild and swap, the index of +# pulled or plugged repositories for its duration, every write of a record for a moment. A writer that finds +# it held waits (saying so after INDEX_LOCK_NOTICE_SECONDS) up to INDEX_LOCK_TIMEOUT seconds, which the +# environment variable changes. +INDEX_LOCK_TIMEOUT_ENV = 'CMETA_INDEX_LOCK_TIMEOUT' +INDEX_LOCK_TIMEOUT = 600 +INDEX_LOCK_NOTICE_SECONDS = 3 + class Repos: """ cMeta repositories manager. @@ -103,17 +111,17 @@ def init( home_path_local = os.path.join(self.home_path, 'local') - # Check if repos do not exist + # Check if repos do not exist (exist_ok: several first runs on one home at once) if not os.path.isdir(self.home_path): self.logger.debug(f"Creating repos directory in {self.home_path} ...") - os.makedirs(self.home_path) + os.makedirs(self.home_path, exist_ok=True) # Check local repo there if not os.path.isdir(home_path_local): self.logger.debug(f"Creating local repo directory in {home_path_local} ...") - os.makedirs(home_path_local) + os.makedirs(home_path_local, exist_ok=True) repo_local_meta_file = os.path.join(home_path_local, self.cfg['repo_meta_desc']) @@ -144,8 +152,18 @@ def init( if r['return']>0: return r if trigger_reindex or not os.path.isdir(self.index_path): - r = self.reindex(con=con, verbose=verbose) - if r['return'] >0: return r + # Under the index lock: a process that arrives while another one builds or swaps the index + # waits for it and then finds the index there, instead of building its own (several first + # runs on one home at once, or a start during the swap of a full reindex) + r = self._lock_index('the first run') + if r['return']>0: return r + index_lock = r['lock'] + try: + if trigger_reindex or not os.path.isdir(self.index_path): + r = self._index_full(con=con, verbose=verbose) + if r['return'] >0: return r + finally: + index_lock.release() return {'return':0} @@ -179,6 +197,18 @@ def add_to_index( Exception: Propagated runtime errors, if any. """ + # Under the index lock: never into an index that a running full reindex is about to replace + r = self._lock_index('the write of a record') + if r['return']>0: return r + index_lock = r['lock'] + try: + return self._add_to_index(cmeta, cmeta_ref_parts, path, original_alias, original_uid, sharding_slices_num, clean_aliases) + finally: + index_lock.release() + + def _add_to_index(self, cmeta, cmeta_ref_parts, path, original_alias, original_uid, sharding_slices_num, clean_aliases): + """`add_to_index` with the index lock held by the caller.""" + category_alias = cmeta_ref_parts['category_alias'].lower() artifact_uid = cmeta_ref_parts['artifact_uid'] artifact_alias = cmeta_ref_parts.get('artifact_alias') @@ -279,8 +309,18 @@ def remove_from_index( Raises: Exception: Propagated runtime errors, if any. """ + r = self._lock_index('the removal of a record') + if r['return']>0: return r + index_lock = r['lock'] + try: + return self._remove_from_index(index_file, artifact_uid, artifact_alias_lowercase) + finally: + index_lock.release() + + def _remove_from_index(self, index_file, artifact_uid, artifact_alias_lowercase): + """`remove_from_index` with the index lock held by the caller.""" r = utils.files.safe_read_file(index_file, lock=True, keep_locked=True, fail_on_error=self.fail_on_error, logger=self.logger) - if r['return']>0: + if r['return']>0: if r['return']!=16: return r index_data = {} @@ -863,8 +903,113 @@ def index( Exception: Propagated runtime errors, if any. """ - if not clean: - return self._index(self.index_path, clean, con, verbose, add_repo_paths, delete_repo_paths) + what = 'the full reindex' if clean else 'the index of the repositories' + note = f'{what} by pid {os.getpid()} on {utils.files._lock_host_name()} since {time.strftime("%Y-%m-%d %H:%M:%S")}' + + r = self._lock_index(what, note=note) + if r['return']>0: return r + index_lock = r['lock'] + try: + if not clean: + return self._index(self.index_path, clean, con, verbose, add_repo_paths, delete_repo_paths) + return self._index_full(con, verbose, add_repo_paths, delete_repo_paths) + finally: + index_lock.release() + + + def _index_lock_file(self): + """The lock file of the index: `/index.lock`, the sidecar of the index folder (the swap of a + full reindex leaves it in place).""" + return utils.files._get_lockfile_path(os.path.normpath(str(self.index_path))) + + + def _lock_index( + self, + why, # What waits, for the notice of a waiter ("the write of a record", "the full reindex" ...). + note = None, # A note left in the lock file for the waiters (who holds it and why). + ): + """ + Take the index lock (`/index.lock`, see INDEX_LOCK_TIMEOUT): a full reindex holds it for + its whole rebuild and swap, the index of pulled or plugged repositories for its duration and + every write of a record for a moment, so that a write never lands in an index that is about + to be replaced and two rebuilds never race for the swap. Readers do not take it. A waiter + says so on stderr after INDEX_LOCK_NOTICE_SECONDS, naming the holder when it left a note, and + gives up after INDEX_LOCK_TIMEOUT seconds (the environment variable INDEX_LOCK_TIMEOUT_ENV + changes that). + + Returns: + dict: {'return': 0, 'lock': } - the caller releases the lock in a `finally` - + or 'return' > 0 and 'error'. + """ + lock_file = self._index_lock_file() + + limit = float(INDEX_LOCK_TIMEOUT) + x = os.environ.get(INDEX_LOCK_TIMEOUT_ENV, '').strip() + if x != '': + try: + limit = max(float(x), 0.0) + except ValueError: + self.logger.warning(f'{INDEX_LOCK_TIMEOUT_ENV}="{x}" is not a number of seconds - {INDEX_LOCK_TIMEOUT} is used') + + lock = utils.files.PathLock(lock_file, logger=self.logger) + first = min(float(INDEX_LOCK_NOTICE_SECONDS), limit) + + try: + lock.acquire(timeout=first, note=note) + except TimeoutError: + holder = utils.files.PathLock.read_note(lock_file) + held_by = f' ({holder})' if holder else '' + message = (f'cMeta: the index in {self.index_path} is locked by another process{held_by} - ' + f'{why} continues when it is released (waiting up to {limit:g} s; {INDEX_LOCK_TIMEOUT_ENV} changes it)') + print(message, file=sys.stderr, flush=True) + self.logger.warning(message) + try: + lock.acquire(timeout=max(limit - first, 0.0), note=note) + except TimeoutError as e: + return {'return':1, 'error': f'the index in {self.index_path} stayed locked by another process{held_by} for {limit:g} s ({e}) - ' + f'set {INDEX_LOCK_TIMEOUT_ENV} to wait longer'} + except Exception as e: + return {'return':1, 'error': f'cannot lock the index in {self.index_path}: {e}'} + + return {'return':0, 'lock':lock} + + + def _remove_index_leftovers( + self, + index_path, # The index folder (normalized). + ): + """Remove every `index.tmp-*` / `index.old-*` sibling of the index folder: with the index lock held + no other rebuild is running, so they are the leftovers of killed or crashed rebuilds.""" + import shutil + + folder = os.path.dirname(index_path) or '.' + base = os.path.basename(index_path) + + try: + names = os.listdir(folder) + except OSError: + return + + for name in names: + if name.startswith(base + '.tmp-') or name.startswith(base + '.old-'): + path = os.path.join(folder, name) + if os.path.isdir(path): + self.logger.debug(f'removing the leftover of an earlier rebuild of the index: {path}') + shutil.rmtree(path, ignore_errors=True) + + + def _index_full( + self, + con = False, # If True, print output to console. + verbose = False, # If True, enable verbose output. + add_repo_paths = [], # Repository paths to add before indexing. + delete_repo_paths = [], # Repository paths to remove before indexing. + ): + """ + The full rebuild of `index` with the index lock held by the caller: the new index is built in + `index.tmp-` and swapped in place of the old one when it is complete; the leftovers of + earlier rebuilds are removed first. + """ import shutil @@ -872,15 +1017,12 @@ def index( index_tmp_path = f'{index_path}.tmp-{os.getpid()}' index_old_path = f'{index_path}.old-{os.getpid()}' - # Leftovers of an earlier run of this process - for path in [index_tmp_path, index_old_path]: - if os.path.isdir(path): - shutil.rmtree(path) + self._remove_index_leftovers(index_path) os.makedirs(index_tmp_path) try: - r = self._index(index_tmp_path, clean, con, verbose, add_repo_paths, delete_repo_paths) + r = self._index(index_tmp_path, True, con, verbose, add_repo_paths, delete_repo_paths) except BaseException: shutil.rmtree(index_tmp_path, ignore_errors=True) raise diff --git a/cmeta/utils/files.py b/cmeta/utils/files.py index 8f00e6f..3023db8 100644 --- a/cmeta/utils/files.py +++ b/cmeta/utils/files.py @@ -49,6 +49,7 @@ LOCK_SOFT_CHECK_SECONDS = 0.25 # how often a waiter on a soft lock reads the file to check for a stale holder LOCK_SOFT_REMOVE_SECONDS = 2.0 # how long a releasing soft lock retries the removal of its file (Windows: a reader may hold it) LOCK_IDENTITY_RETRIES = 50 # consecutive identity mismatches after which a path is locked with the soft lock +LOCK_NOTE_MAX = 512 # bytes of the note a holder may leave in its lock file for the message of a waiter ERROR_CODE_FILE_NOT_FOUND = 16 RETRY_DELAY = 0.1 RETRY_NOT_FOUND_FILE = 10 @@ -239,6 +240,7 @@ def __del__(self): def acquire( self, timeout: float = 3, # Seconds to wait for the lock (0: one try). + note: str = None, # A short note left in the lock file for the message of a waiter (see read_note). ): """Take the lock or raise TimeoutError; returns self.""" timeout = max(float(timeout), 0.0) @@ -254,6 +256,8 @@ def acquire( self._thread_lock = thread_lock self.is_locked = True self.logger.debug(f"utils.files.PathLock - acquired {self.lock_file} ({self.mode})") + if note is not None: + self.write_note(note) return self remaining = deadline - time.monotonic() @@ -303,6 +307,48 @@ def release(self): if thread_lock is not None: thread_lock.release() + def write_note(self, note): + """Leave a short note in the held lock file - who holds it and why - for the message of a waiter + (`read_note`). Best effort: nothing happens when it cannot be written. The note starts after a first + byte that stays free: on Windows that byte is the locked one, which another process cannot read. Not + in soft mode, where the file already holds `pid host time`.""" + if not self.is_locked or self._fd is None or self.mode == 'soft': + return + data = b'\n' + str(note).encode('utf-8', 'replace')[:LOCK_NOTE_MAX] + b'\n' + try: + os.ftruncate(self._fd, 0) + os.lseek(self._fd, 0, os.SEEK_SET) + os.write(self._fd, data) + except OSError: + pass + finally: + try: + os.lseek(self._fd, 0, os.SEEK_SET) # the Windows byte lock is addressed by the file position + except OSError: + pass + + @staticmethod + def read_note(lock_file): + """The note the holder of `lock_file` left (`write_note`), or the `pid host time` of a soft lock, + or '' when there is none or the file cannot be read.""" + try: + fd = os.open(lock_file, os.O_RDONLY | getattr(os, 'O_BINARY', 0)) + except OSError: + return '' + try: + try: + data = os.read(fd, LOCK_NOTE_MAX + 8) + except OSError: + # Windows: the first byte is locked by the holder - read from the second one + os.lseek(fd, 1, os.SEEK_SET) + data = os.read(fd, LOCK_NOTE_MAX + 8) + except OSError: + return '' + finally: + os.close(fd) + text = data.decode('utf-8', 'replace').strip() + return text.splitlines()[0].strip() if text else '' + # ---- one attempt ---------------------------------------------------------------------- def _try(self): diff --git a/cmeta/version.py b/cmeta/version.py index c3b0502..9289284 100644 --- a/cmeta/version.py +++ b/cmeta/version.py @@ -6,5 +6,5 @@ See the cMeta COPYRIGHT and LICENSE files in the project root for details. """ -__version__ = "0.34.0" +__version__ = "0.34.0.1" __release_date__ = "2026-10-09" diff --git a/docs/async-and-concurrency.md b/docs/async-and-concurrency.md index 5f9b3de..865c56c 100644 --- a/docs/async-and-concurrency.md +++ b/docs/async-and-concurrency.md @@ -249,6 +249,25 @@ with a delay that doubles from 1 ms to 25 ms until the timeout. A folder used by two operating systems at once (a Windows tree plugged from WSL) is best effort: each side locks with its own mechanism. +### The index lock + +A full reindex (`cx --reindex`, or the first run on a home without an index) +builds the new index in `index.tmp-` and swaps it in when it is complete. +So that a record written meanwhile is not lost in the swap, the lock file +`/index.lock` (the sidecar of the index folder, kept through the +swap) is held by the full reindex for its whole rebuild and swap, by the +incremental index of `pull` / `plug` / `repo reindex` for its duration, and by +every write of a record (`create`, `update`, `delete`, `reindex` of an artifact) +for a moment. A writer that arrives during a reindex waits for it, prints a +notice on stderr after 3 seconds naming the reindexing process (which leaves a +note in the lock file), and gives up with an error after `CMETA_INDEX_LOCK_TIMEOUT` +seconds (600 by default); otherwise it then writes into the new index. Two +reindexes run one after the other; several processes starting on a fresh home +at once build one index. Readers never take the lock, so lookups cost what they +did; a write costs one more lock acquisition (microseconds on Linux, a fraction +of a millisecond on Windows). Under the same lock a reindex removes the +`index.tmp-*` / `index.old-*` leftovers of killed rebuilds. + ### Measured behaviour Ten `cx note add` processes run simultaneously against one `` all @@ -282,8 +301,10 @@ files; `tests/benchmarks/benchmark_engine.py` measures the cost. - **Separate ``s remove sharing entirely.** For fully independent parallel jobs, give each one its own home (`--home=` or `CMETA_HOME`) — see [using-cmeta.md §7.4](using-cmeta.md#74-picking-cmeta_home-env-vars--per-project-collections). -- **Reindex is a whole-home operation.** Avoid running `cx --reindex` while - other processes are writing to the same home. +- **Reindex is a whole-home operation.** Writes made while `cx --reindex` runs + wait for it (see *The index lock* above) and land in the new index; a very + long reindex (a huge home on a slow disk) makes them wait as long, up to + `CMETA_INDEX_LOCK_TIMEOUT` seconds. --- diff --git a/docs/using-cmeta.md b/docs/using-cmeta.md index 131d0bc..6ebe6ed 100644 --- a/docs/using-cmeta.md +++ b/docs/using-cmeta.md @@ -34,7 +34,8 @@ If you haven't installed cMeta yet, start with Environment variables worth knowing: `CMETA_HOME` (state directory, default under user home), `CMETA_DEBUG=1`, `CMETA_LOG=DEBUG|INFO`, `CMETA_LOG_FILE=`, `CMETA_VERBOSE=yes`, -`CMETA_FAIL_ON_ERROR=yes`. +`CMETA_FAIL_ON_ERROR=yes`, `CMETA_INDEX_LOCK_TIMEOUT=` (how long a +write to the index waits for a running `cx --reindex`; 600). --- diff --git a/tests/internal_repo_tests/test_integration_index_guard.py b/tests/internal_repo_tests/test_integration_index_guard.py new file mode 100644 index 0000000..fb16afe --- /dev/null +++ b/tests/internal_repo_tests/test_integration_index_guard.py @@ -0,0 +1,228 @@ +""" +Integration tests of the index lock (`/index.lock`): a full reindex holds it for its whole rebuild +and swap, the index of pulled or plugged repositories for its duration, and every write of a record for +a moment - so the writes made while a reindex runs wait for it and land in the new index instead of in +the one it replaces, two rebuilds run one after the other, several first runs on a fresh home build one +index, the leftovers of killed rebuilds are removed, and a waiter says so and gives up after +CMETA_INDEX_LOCK_TIMEOUT seconds. + +Each test runs against a fresh in tmp_path so it does not touch the user's real cMeta +state. The reindex of another process is paused by a hook on `Repos._index`: a marker file says that +it holds the lock, a go file lets it continue. + +cMeta author and developer: (C) 2025-2026 Grigori Fursin + +See the cMeta COPYRIGHT and LICENSE files in the project root for details. +""" + +import os +import pickle +import subprocess +import sys +import time + +import pytest + +from cmeta import CMeta +from cmeta import repos as cmeta_repos +from cmeta.utils import files + + +@pytest.fixture() +def cm(tmp_path, monkeypatch): + for var in ('CMETA_HOME', 'CMETA_HOME2', 'VIRTUAL_ENV', 'CONDA_PREFIX', + 'CMETA_DEBUG', 'CMETA_VERBOSE', 'CMETA_FAIL_ON_ERROR', 'CMETA_INDEX_LOCK_TIMEOUT'): + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv('CMETA_HOME', str(tmp_path)) + monkeypatch.setattr(cmeta_repos, '_migrated_notices', set()) + return CMeta(home=str(tmp_path)) + + +def make(cm, alias, **meta): + r = cm.access({'category': 'log', 'command': 'create', 'arg1': f'local:{alias}', 'meta': meta, 'con': False}) + assert r['return'] == 0, r.get('error') + return r['meta']['artifact'] + + +def env_for(home, **extra): + env = {k: v for k, v in os.environ.items() if not k.startswith('CMETA_') and k not in ('VIRTUAL_ENV', 'CONDA_PREFIX')} + env['CMETA_HOME'] = str(home) + env['PYTHONIOENCODING'] = 'utf-8' + env.update(extra) + return env + + +def start(args, home, **extra): + """A cMeta command line (cx ) in another process, with a null input; its output is read by finish().""" + return subprocess.Popen([sys.executable, '-m', 'cmeta'] + list(args), env=env_for(home, **extra), + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + + +def finish(proc, timeout=300): + out, _ = proc.communicate(timeout=timeout) + return proc.returncode, out.decode('utf-8', 'replace') + + +PAUSED_REINDEX = r''' +import os, sys, time +home, marker, go = sys.argv[1:4] +from cmeta import CMeta +cm = CMeta(home=home) +original = cm.repos._index +def paused(*args, **kwargs): + with open(marker, 'w') as f: + f.write(str(os.getpid())) + while not os.path.exists(go): + time.sleep(0.01) + return original(*args, **kwargs) +cm.repos._index = paused +r = cm.repos.reindex() +print('REINDEX', r['return'], r.get('error', ''), flush=True) +''' + + +def start_paused_reindex(tmp_path): + """A full reindex in another process that holds the index lock and waits for the go file.""" + marker = tmp_path / 'reindex-holds-the-lock' + go = tmp_path / 'reindex-go' + proc = subprocess.Popen([sys.executable, '-c', PAUSED_REINDEX, str(tmp_path), str(marker), str(go)], + env=env_for(tmp_path), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + deadline = time.time() + 60 + while not marker.exists(): + if proc.poll() is not None: + raise AssertionError('the reindex ended before it held the lock: ' + finish(proc)[1]) + if time.time() > deadline: + proc.kill() + raise AssertionError('the paused reindex did not take the lock in 60 s') + time.sleep(0.01) + return proc, go + + +def index_aliases(tmp_path, category='log'): + """(the lowercase aliases of the index, {uid: path}) of a category as the pickle on disk holds them.""" + with open(tmp_path / 'index' / f'{category}.pkl', 'rb') as f: + data = pickle.load(f) + return set(data['lowercase_aliases']), {uid: rec['path'] for uid, rec in data['uids'].items()} + + +def folders(home, category='log'): + base = home / 'repos' / 'local' / category + return {p.name.lower() for p in base.iterdir() if p.is_dir()} if base.is_dir() else set() + + +def leftovers(home): + return sorted(p.name for p in home.iterdir() if p.name.startswith(('index.tmp-', 'index.old-', 'index.lock'))) + + +def test_writes_during_a_full_reindex_wait_for_it_and_land_in_the_new_index(cm, tmp_path): + for i in range(12): + make(cm, f'kept-{i}', n=i) + for i in range(3): + make(cm, f'doomed-{i}') + + reindex, go = start_paused_reindex(tmp_path) + + writers = { + 'create': [start(['log', 'create', f'born-{i}'], tmp_path) for i in range(3)], + 'delete': [start(['log', 'delete', f'doomed-{i}', '-f'], tmp_path) for i in range(3)], + 'update': [start(['log', 'update', 'kept-0', '--new_tags=touched'], tmp_path)], + 'reindex one': [start(['log', 'reindex', 'kept-1'], tmp_path)], + 'repo reindex': [start(['repo', 'reindex', 'local'], tmp_path)], + } + every = [p for group in writers.values() for p in group] + + # While the reindex holds the lock none of the writers gets through + time.sleep(2.0) + assert all(p.poll() is None for p in every), 'a writer finished while the reindex held the index lock' + + go.touch() + rc, out = finish(reindex) + assert rc == 0 and 'REINDEX 0' in out, out + + for kind, group in writers.items(): + for p in group: + rc, out = finish(p) + assert rc == 0, f'{kind}: {out}' + + # The index equals the disk: the creations are in, the deletions are out, the update is in + aliases, paths = index_aliases(tmp_path) + expected = {f'kept-{i}' for i in range(12)} | {f'born-{i}' for i in range(3)} + assert folders(tmp_path) == expected + assert aliases == expected + assert all(os.path.isdir(p) for p in paths.values()) + r = cm.access({'category': 'log', 'command': 'find', 'arg1': 'kept-0', 'con': False}) + assert r['return'] == 0 and 'touched' in r['artifacts'][0]['cmeta'].get('tags', []), r + assert leftovers(tmp_path) == [] + + +def test_two_full_reindexes_at_once_run_one_after_the_other(cm, tmp_path): + for i in range(10): + make(cm, f'a-{i}') + + procs = [start(['--reindex'], tmp_path) for _ in range(2)] + for p in procs: + rc, out = finish(p) + assert rc == 0, out + + aliases, paths = index_aliases(tmp_path) + assert aliases == folders(tmp_path) == {f'a-{i}' for i in range(10)} + assert all(os.path.isdir(p) for p in paths.values()) + assert leftovers(tmp_path) == [] + + +def test_several_first_runs_on_a_fresh_home_build_one_index(tmp_path): + home = tmp_path / 'fresh' + + procs = [start(['repo', 'find', 'local'], home) for _ in range(4)] + for p in procs: + rc, out = finish(p) + assert rc == 0, out + + assert (home / 'repos.json').is_file() + assert (home / 'index' / 'repo.pkl').is_file() + assert leftovers(home) == [] + + +def test_the_leftovers_of_killed_rebuilds_are_removed_by_the_next_reindex(cm, tmp_path): + make(cm, 'one') + for name in ('index.tmp-424242', 'index.old-424243'): + d = tmp_path / name + d.mkdir() + (d / 'log.pkl').write_bytes(b'leftover') + + r = cm.repos.reindex() + assert r['return'] == 0, r.get('error') + assert leftovers(tmp_path) == [] + assert (tmp_path / 'index' / 'repo.pkl').is_file() + assert index_aliases(tmp_path)[0] == {'one'} + + +def test_a_writer_waiting_for_a_reindex_says_so_and_gives_up_after_the_limit(cm, tmp_path): + make(cm, 'present') + reindex, go = start_paused_reindex(tmp_path) + try: + t0 = time.time() + writer = start(['log', 'create', 'late'], tmp_path, CMETA_INDEX_LOCK_TIMEOUT='5') + rc, out = finish(writer, timeout=120) + elapsed = time.time() - t0 + finally: + go.touch() + rc_reindex, out_reindex = finish(reindex) + + assert rc != 0, out + assert 'is locked by another process' in out and 'the full reindex by pid' in out, out + assert 'stayed locked by another process' in out and 'CMETA_INDEX_LOCK_TIMEOUT' in out, out + assert 4.5 <= elapsed < 60, elapsed + assert rc_reindex == 0 and 'REINDEX 0' in out_reindex, out_reindex + + +def test_the_note_of_a_lock_file_is_readable_by_a_waiter(tmp_path): + lock_file = str(tmp_path / 'x.lock') + lock = files.PathLock(lock_file).acquire(note='the full reindex by pid 7') + try: + assert lock.is_locked + assert files.PathLock.read_note(lock_file) == 'the full reindex by pid 7' + finally: + lock.release() + assert not os.path.exists(lock_file) + assert files.PathLock.read_note(lock_file) == '' From e1632f8d3b84de0bfc66604c208712c19ec38d22 Mon Sep 17 00:00:00 2001 From: Grigori Fursin Date: Fri, 9 Oct 2026 14:59:32 +0200 Subject: [PATCH 3/4] The write side: a writer waits 30 s for a busy lock and says who holds it, structured files are written atomically, a create is one step under the index lock, repos.json is locked by the incremental index, the reader cache keys on time, size and inode (0.34.0.1) - CMETA_LOCK_TIMEOUT (30 s) for the lock of a file: a notice on stderr after 3 s naming the holder's note, then an error naming it; timeout=None = the engine's defaults in safe_read_file, safe_write_file, safe_delete_directory, lock_path; the no-lock read retry stays 3 s; lock_timeout() and _acquire_with_notice() shared with the index lock - safe_write_file(atomic=None): JSON, YAML and pickle written next to the target and put in place with one replace (_install_file: the mode kept, a symbolic link written through, a read-only target refused); a failed write removes its temporary; text stays in place - create_ makes the folder, the meta and the record under the index lock (Repos.lock_index, add_to_index(index_lock=)); several creates of one artifact end with one artifact and "already exists"; a create whose record cannot be written removes what it made (_undo_create) - _index reads and rewrites repos.json under its lock; a new category index file is written atomically too - safe_read_file_via_cache keyed by (st_mtime_ns, st_size, st_ino), its dict updated under a lock, one stat per lookup; a file read before that is missing for a moment (a replace on a mount whose rename is not atomic, the swap of a full reindex) is retried three times before "not found" - _replace_file retried on EACCES/EPERM/EBUSY with a delay doubling 1 ms -> 100 ms until the lock timeout; a target that cannot be written fails at once - Repos.find hands out one-level copies of the records it returns (the record, its cmeta, its tags), made after its tag and match filters; find_in_index is unchanged - index files written with pickle protocol 4 - tests/core_tests/test_utils_files_writes.py, tests/internal_repo_tests/test_integration_create_guard.py, the index guard test checks that a create waits before it writes its folder - docs/async-and-concurrency.md, AGENTS.md and docs/using-cmeta.md env lists, CHANGELOG Co-Authored-By: Claude Fable 5.1 Signed-off-by: Grigori Fursin --- AGENTS.md | 3 +- CHANGELOG.md | 43 +++ cmeta/category_api_v1.py | 216 +++++++---- cmeta/repos.py | 99 +++-- cmeta/utils/files.py | 363 +++++++++++++----- docs/async-and-concurrency.md | 56 ++- docs/using-cmeta.md | 4 +- tests/core_tests/test_utils_files_writes.py | 309 +++++++++++++++ .../test_integration_create_guard.py | 186 +++++++++ .../test_integration_index_guard.py | 6 +- 10 files changed, 1060 insertions(+), 225 deletions(-) create mode 100644 tests/core_tests/test_utils_files_writes.py create mode 100644 tests/internal_repo_tests/test_integration_create_guard.py diff --git a/AGENTS.md b/AGENTS.md index abff02d..8d087c5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -64,7 +64,8 @@ tabulate, tqdm, packaging, psutil. (File locks are the engine's own `CMETA_HOME`, `CMETA_DEBUG=1`, `CMETA_LOG=DEBUG|INFO`, `CMETA_LOG_FILE=`, `CMETA_FAIL_ON_ERROR=yes`, `CMETA_VERBOSE=yes`, `CMETA_INTERNAL_REPO_PATH`, `CMETA_PIP_INSTALL_ARGS`, `CMETA_INDEX_LOCK_TIMEOUT` (seconds a write to the index waits for a running -reindex, 600), and the provenance stamped on new artifacts: `CMETA_AUTHORS`, `CMETA_COPYRIGHT`, +reindex, 600), `CMETA_LOCK_TIMEOUT` (seconds a writer waits for the lock of a file, 30), and the +provenance stamped on new artifacts: `CMETA_AUTHORS`, `CMETA_COPYRIGHT`, `CMETA_GENERATOR` (JSON: how the artifact was made; with the `artifact_defaults` of a repository's `_cmr.yaml` - docs/using-cmeta.md §7.6). diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b49c59..d78c386 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,49 @@ All notable changes to cMeta are documented here, newest first. notice and the timeout, the note). With it, the first run creates the home and its local repository folder with `exist_ok`: two first runs on one home at once no longer fail each other with "cannot create a file when that file already exists". +- **The write side made whole: a writer waits 30 s for a busy lock and says who holds it, every + structured file is written atomically, a create is one step under the index lock, the index of + repositories holds the lock of `repos.json`, the reader cache keys on time, size and inode, a replace + is retried until the lock timeout, `find` hands out copies, index files use pickle protocol 4.** + - A writer that found the lock of a file held (an index file, a meta, `repos.json`) waited 3 s and + failed; under load (server workers, a test matrix) a `create` had written its meta and then failed to + index it, leaving a folder the index did not know. A writer now waits up to `CMETA_LOCK_TIMEOUT` + seconds (30), prints a notice on stderr after 3 s naming the holder when it left a note in the lock + file, and then fails with an error that names the holder and the variable. `timeout=None` (the new + default of `safe_read_file`, `safe_write_file`, `safe_delete_directory`, `lock_path`) means the + engine's timeouts; an explicit value is used as before; without a lock, a read that fails is still + retried for 3 s. `cmeta.utils.files.lock_timeout()` reads the setting. + - `safe_write_file` writes a JSON, YAML or pickle file atomically by default (`.tmp` + one + replace; text stays in place): a new artifact's meta, the first run's `repos.json` and local + repository meta, a new category index file and the files of the internal categories can no longer be + seen half-written or left broken by a crash (a broken meta made the next reindex skip the artifact). + A failed write removes its temporary; the mode of the target is kept; a symbolic link is written + through; a read-only target is refused as before. + - `create` makes the folder, the meta and the record under the index lock (`Repos.lock_index`, + `add_to_index(index_lock=...)`): a full reindex never sees a half-made artifact, several creates of + one artifact at once end with one artifact and "already exists" errors (return code 8), and a create + whose record cannot be written removes what it made. + - The incremental index (`pull`, `plug`, `unplug`, `repo reindex`) reads and rewrites `repos.json` + under its lock: a plug or unplug of another process at the same moment is neither lost nor read + half-written. + - The reader cache of the index (`safe_read_file_via_cache`) is keyed by the modification time in + nanoseconds, the size and the inode instead of the modification time in seconds: two writes within + one tick of a coarse file system (FAT, exFAT, some network mounts) no longer leave a long-running + process with a stale index; an index file read before that is missing for a moment (a replace on a + mount whose rename is not atomic, such as 9p, or the swap of a full reindex) is retried a few times + (75 ms at most) before "not found" - a file never seen is still not retried; its dict is updated + under a lock for the threads of one process; two `stat`s fewer per lookup. + - A replace whose target is held open (Windows: a reader of an index file, a scanner) is retried with + a delay that doubles from 1 ms to 100 ms until the lock timeout, instead of 10 times 0.1 s; a target + that cannot be written at all fails at once. + - `find` hands out one-level copies of the records it returns (the record, its `cmeta`, its `tags`; + made after its tag and match filters, so only what is returned pays for it): a caller that edits + what it gets (the aops selection adds a tag) no longer edits the cached index of its process. Deeper + values, and the records of the lower-level `find_in_index`, are shared and must be left as they are. + - Index files are written with pickle protocol 4, readable by every supported Python whatever the + default of the Python that writes. Tests: `tests/core_tests/test_utils_files_writes.py`, + `tests/internal_repo_tests/test_integration_create_guard.py`; the index guard test now checks that a + create waits before it writes its folder. ## 0.34.0 - **The engine's own path lock replaces the `filelock` library: two processes can no longer hold one diff --git a/cmeta/category_api_v1.py b/cmeta/category_api_v1.py index 1d4e595..784dec9 100644 --- a/cmeta/category_api_v1.py +++ b/cmeta/category_api_v1.py @@ -1076,110 +1076,136 @@ def create_( # The meta file of an artifact that is indexed as it is (made or copied by hand) existing_meta_file = None - if not virtual: - if os.path.isdir(artifact_path): - if os.path.isfile(cmeta_filename_json) or os.path.isfile(cmeta_filename_yaml): - if index: - if os.path.isfile(cmeta_filename_json): - f = cmeta_filename_json - else: - f = cmeta_filename_yaml - yaml = True - - # Do not use lock since it's not in the system and not in the index - r = utils.files.safe_read_file(f, fail_on_error=self.fail_on_error, logger=self.logger) - if r['return']>0: return r - - meta = r['data'] - existing_meta_file = f + # The folder, the meta and the record of a new artifact are made under the index lock: a full + # reindex running meanwhile never sees a half-made artifact, two creates of one artifact at once + # end with one artifact (the second finds the folder of the first), and a create whose record + # cannot be written removes what it made, so that nothing is left unindexed + index_lock = None + if not no_index: + r = self.cm.repos.lock_index('the creation of an artifact') + if r['return']>0: return r + index_lock = r['lock'] + + disk_path = artifact_path # where the folder and the meta go (`path` may replace artifact_path for the index) + made_folder = False # the create made the folder itself + written_meta_file = None # the meta file the create wrote + done = False + + try: + if not virtual: + if os.path.isdir(artifact_path): + if os.path.isfile(cmeta_filename_json) or os.path.isfile(cmeta_filename_yaml): + if index: + if os.path.isfile(cmeta_filename_json): + f = cmeta_filename_json + else: + f = cmeta_filename_yaml + yaml = True + + # Do not use lock since it's not in the system and not in the index + r = utils.files.safe_read_file(f, fail_on_error=self.fail_on_error, logger=self.logger) + if r['return']>0: return r + + meta = r['data'] + existing_meta_file = f + + # Index the artifact under the UID of its meta (an "alias,UID" too), + # not a new one, or alias,UID references to it would not resolve + meta_uid = str(meta.get('artifact') or '').split(',')[-1].strip() + if meta_uid and utils.names.is_valid_cmeta_uid(meta_uid): + artifact_uid = meta_uid - # Index the artifact under the UID of its meta (an "alias,UID" too), - # not a new one, or alias,UID references to it would not resolve - meta_uid = str(meta.get('artifact') or '').split(',')[-1].strip() - if meta_uid and utils.names.is_valid_cmeta_uid(meta_uid): - artifact_uid = meta_uid + else: + return {'return':8, 'error':f'artifact already exists in "{artifact_path}"'} + else: + made_folder = True - else: - return {'return':8, 'error':f'artifact already exists in "{artifact_path}"'} + os.makedirs(artifact_path, exist_ok=True) - os.makedirs(artifact_path, exist_ok=True) + # Prepare meta + cmeta = copy.deepcopy(meta) - # Prepare meta - cmeta = copy.deepcopy(meta) + if 'artifact' not in cmeta: + cmeta['artifact'] = artifact_uid - if 'artifact' not in cmeta: - cmeta['artifact'] = artifact_uid + if 'category' not in cmeta: + r = utils.names.restore_cmeta_name(category, key='artifact', fail_on_error = self.fail_on_error) + if r['return'] >0: return r - if 'category' not in cmeta: - r = utils.names.restore_cmeta_name(category, key='artifact', fail_on_error = self.fail_on_error) - if r['return'] >0: return r + cmeta['category'] = r['name'] - cmeta['category'] = r['name'] + if tags is not None: + meta_tags = cmeta.get('tags', []) - if tags is not None: - meta_tags = cmeta.get('tags', []) + r = utils.common.normalize_tags(tags) + if r['return']>0: return r - r = utils.common.normalize_tags(tags) - if r['return']>0: return r - - for t in r['tags']: - if t not in meta_tags: - meta_tags.append(t) + for t in r['tags']: + if t not in meta_tags: + meta_tags.append(t) - cmeta['tags'] = meta_tags + cmeta['tags'] = meta_tags - if existing_meta_file is None: - from datetime import datetime, timezone + if existing_meta_file is None: + from datetime import datetime, timezone - if 'creation_timestamp' not in cmeta: - cmeta['creation_timestamp'] = datetime.now(timezone.utc).isoformat() - else: - cmeta['last_update_timestamp'] = datetime.now(timezone.utc).isoformat() + if 'creation_timestamp' not in cmeta: + cmeta['creation_timestamp'] = datetime.now(timezone.utc).isoformat() + else: + cmeta['last_update_timestamp'] = datetime.now(timezone.utc).isoformat() + + # Authors, copyright and generator: the meta, the environment and the repository defaults + apply_artifact_defaults(cmeta, repo_meta=repo_cmeta, cfg=self.cm.cfg, creating=True) + + # Save meta + if not virtual: + if existing_meta_file is not None: + # Indexing leaves the meta file as it was written; only what the request adds (the tags, + # an identity key the file lacks) is added in place, without rewriting the rest + if cmeta != meta: + r = utils.files.safe_write_file(existing_meta_file, data=cmeta, fail_on_error = self.fail_on_error, logger=self.logger, preserve=True) + if r['return']>0: return r - # Authors, copyright and generator: the meta, the environment and the repository defaults - apply_artifact_defaults(cmeta, repo_meta=repo_cmeta, cfg=self.cm.cfg, creating=True) + else: + tmp_cmeta_filename = cmeta_filename_yaml if yaml else cmeta_filename_json + written_meta_file = tmp_cmeta_filename - # Save meta - if not virtual: - if existing_meta_file is not None: - # Indexing leaves the meta file as it was written; only what the request adds (the tags, - # an identity key the file lacks) is added in place, without rewriting the rest - if cmeta != meta: - r = utils.files.safe_write_file(existing_meta_file, data=cmeta, fail_on_error = self.fail_on_error, logger=self.logger, preserve=True) + # A new YAML meta is written in the keep style (the key order of YAML_META_KEY_ORDER, + # nothing folded, lists indented), the style `update` keeps; a JSON meta is sorted as before + r = utils.files.safe_write_file(tmp_cmeta_filename, data=cmeta, fail_on_error = self.fail_on_error, keep=True) if r['return']>0: return r - else: - tmp_cmeta_filename = cmeta_filename_yaml if yaml else cmeta_filename_json - - # A new YAML meta is written in the keep style (the key order of YAML_META_KEY_ORDER, - # nothing folded, lists indented), the style `update` keeps; a JSON meta is sorted as before - r = utils.files.safe_write_file(tmp_cmeta_filename, data=cmeta, fail_on_error = self.fail_on_error, keep=True) - if r['return']>0: return r + if path is not None: + artifact_path = path - if path is not None: - artifact_path = path + # Update index + if not no_index: + cmeta_ref_parts = {} - # Update index - if not no_index: - cmeta_ref_parts = {} + if artifact_alias is not None: + cmeta_ref_parts['artifact_alias'] = artifact_alias + artifact_alias_lowercase = artifact_alias.lower() + if artifact_alias_lowercase != artifact_alias: + cmeta_ref_parts['artifact_alias_lowercase'] = artifact_alias_lowercase - if artifact_alias is not None: - cmeta_ref_parts['artifact_alias'] = artifact_alias - artifact_alias_lowercase = artifact_alias.lower() - if artifact_alias_lowercase != artifact_alias: - cmeta_ref_parts['artifact_alias_lowercase'] = artifact_alias_lowercase + cmeta_ref_parts['artifact_uid'] = artifact_uid - cmeta_ref_parts['artifact_uid'] = artifact_uid + cmeta_ref_parts['category_alias'] = category_alias + cmeta_ref_parts['category_uid'] = category_uid - cmeta_ref_parts['category_alias'] = category_alias - cmeta_ref_parts['category_uid'] = category_uid + cmeta_ref_parts['repo_alias'] = artifact_repo_alias + cmeta_ref_parts['repo_uid'] = artifact_repo_uid - cmeta_ref_parts['repo_alias'] = artifact_repo_alias - cmeta_ref_parts['repo_uid'] = artifact_repo_uid + r = self.cm.repos.add_to_index(cmeta, cmeta_ref_parts, artifact_path, sharding_slices_num = sharding_slices_num, index_lock = index_lock) + if r['return']>0: return r + done = True - r = self.cm.repos.add_to_index(cmeta, cmeta_ref_parts, artifact_path, sharding_slices_num = sharding_slices_num) - if r['return']>0: return r + finally: + if not done and not virtual: + self._undo_create(disk_path, made_folder, written_meta_file, sharding_slices) + if index_lock is not None: + index_lock.release() # Print artifact path if con: @@ -1188,6 +1214,36 @@ def create_( return {'return':0, 'path':artifact_path, 'meta':cmeta} + ############################################################ + def _undo_create( + self, + artifact_path: str, # The folder of the artifact that was being created. + made_folder: bool, # True when the create made the folder itself: it goes with everything in it. + meta_file: str, # The meta file the create wrote, or None. + sharding_slices, # The sharding of the category, to remove the empty parent folders too. + ): + """ + Remove what a create that failed made on disk - the folder when the create made it, else only + the meta it wrote - so that a create whose record could not be written leaves nothing unindexed. + + Args: + artifact_path (str): The folder of the artifact that was being created. + made_folder (bool): True when the create made the folder itself. + meta_file (str | None): The meta file the create wrote. + sharding_slices (list | None): The sharding of the category. + """ + import shutil + + if made_folder: + shutil.rmtree(artifact_path, ignore_errors=True) + utils.files.safe_delete_directory_if_empty_with_sharding(artifact_path, sharding_slices) + + elif meta_file is not None: + try: + os.remove(meta_file) + except OSError: + pass + ############################################################ def move_( self, diff --git a/cmeta/repos.py b/cmeta/repos.py index 3faa117..7483f95 100644 --- a/cmeta/repos.py +++ b/cmeta/repos.py @@ -177,9 +177,10 @@ def add_to_index( original_uid = None, # Previous artifact UID before update. sharding_slices_num = 0, # Number of sharding slices clean_aliases = False, # If True, drop every other alias of this UID from the alias index (a folder renamed by hand), in the same locked write. + index_lock = None, # The index lock when the caller holds it already (lock_index): the record is written under it, without taking it again. ): """ - Insert or update an artifact record in the category index. + Insert or update an artifact record in the category index (under the index lock, see `_lock_index`). Args: cmeta: Artifact metadata dictionary. @@ -191,6 +192,8 @@ def add_to_index( clean_aliases (bool): If True, every alias of the alias index that still lists this UID and is not the alias of the record is dropped (the stale aliases a folder renamed by hand leaves behind), in the same locked write. Aliases of other UIDs are not touched. + index_lock: The index lock when the caller holds it already (`lock_index`): the record is + written under it, without taking it again (the lock is not re-entrant). Returns: dict: Operation result, with 'dropped_aliases': the (lowercase) aliases dropped by `clean_aliases`. Raises: @@ -198,6 +201,9 @@ def add_to_index( """ # Under the index lock: never into an index that a running full reindex is about to replace + if index_lock is not None: + return self._add_to_index(cmeta, cmeta_ref_parts, path, original_alias, original_uid, sharding_slices_num, clean_aliases) + r = self._lock_index('the write of a record') if r['return']>0: return r index_lock = r['lock'] @@ -467,6 +473,8 @@ def find_in_index( if artifact_uid not in index[self.KEY_INDEX_UIDS]: return _error(f'corrupted index for {category_alias} UID "{artifact_uid}"', 1, None, self.fail_on_error) + # A shallow copy of the record: its 'cmeta' is the cached object itself - do not change + # it (`find` copies it for the records it returns) artifact = index[self.KEY_INDEX_UIDS][artifact_uid].copy() if add_index_file: @@ -668,15 +676,19 @@ def find( category_cmeta = category['cmeta'] + from_index = False + if category_cmeta.get('no_index', False) and not skip_non_indexed: r = self.find_in_file_system(category_cmeta, category_alias, category_uid, artifact_alias, artifact_uid, repo_uids = artifact_repo_artifacts) - if r['return'] >0: + if r['return'] >0: if r['return'] == 16: # If index not found continue return r else: + from_index = True + r = self.find_in_index(category_alias, category_uid, artifact_alias, artifact_uid, repos = artifact_repo_artifacts, add_index_file = add_index_file, skip_uids=skip_uids) if r['return'] >0: if r['return'] == 16: @@ -759,8 +771,22 @@ def find( add_artifacts = add_artifacts2 - # Adding artifacts - artifacts.extend(add_artifacts) + # Adding artifacts: those from the index get their own copy of the meta and of its tags + # (one level: a caller that edits them - the aops selection adds a tag - edits its copy, + # not the cached index of this process; deeper values are shared and must be left as + # they are). Only the records returned pay for it, after the filters above. + if from_index: + for a in add_artifacts: + cmeta = a['cmeta'] + if type(cmeta) is dict: + cmeta = dict(cmeta) + ctags = cmeta.get('tags') + if type(ctags) is list: + cmeta['tags'] = ctags[:] + a['cmeta'] = cmeta + artifacts.append(a) + else: + artifacts.extend(add_artifacts) if len(artifacts) == 0: x_artifact_alias = "artifacts" if artifact_alias == '' or artifact_alias == None else f'"{artifact_alias}"' @@ -942,38 +968,36 @@ def _lock_index( or 'return' > 0 and 'error'. """ lock_file = self._index_lock_file() - - limit = float(INDEX_LOCK_TIMEOUT) - x = os.environ.get(INDEX_LOCK_TIMEOUT_ENV, '').strip() - if x != '': - try: - limit = max(float(x), 0.0) - except ValueError: - self.logger.warning(f'{INDEX_LOCK_TIMEOUT_ENV}="{x}" is not a number of seconds - {INDEX_LOCK_TIMEOUT} is used') - + limit = utils.files.lock_timeout(INDEX_LOCK_TIMEOUT, INDEX_LOCK_TIMEOUT_ENV, logger=self.logger) lock = utils.files.PathLock(lock_file, logger=self.logger) - first = min(float(INDEX_LOCK_NOTICE_SECONDS), limit) try: - lock.acquire(timeout=first, note=note) - except TimeoutError: - holder = utils.files.PathLock.read_note(lock_file) - held_by = f' ({holder})' if holder else '' - message = (f'cMeta: the index in {self.index_path} is locked by another process{held_by} - ' - f'{why} continues when it is released (waiting up to {limit:g} s; {INDEX_LOCK_TIMEOUT_ENV} changes it)') - print(message, file=sys.stderr, flush=True) - self.logger.warning(message) - try: - lock.acquire(timeout=max(limit - first, 0.0), note=note) - except TimeoutError as e: - return {'return':1, 'error': f'the index in {self.index_path} stayed locked by another process{held_by} for {limit:g} s ({e}) - ' - f'set {INDEX_LOCK_TIMEOUT_ENV} to wait longer'} + utils.files._acquire_with_notice(lock, limit, f'the index in {self.index_path}', why, INDEX_LOCK_TIMEOUT_ENV, note=note, logger=self.logger) + except TimeoutError as e: + return {'return':1, 'error': str(e)} except Exception as e: return {'return':1, 'error': f'cannot lock the index in {self.index_path}: {e}'} return {'return':0, 'lock':lock} + def lock_index( + self, + why, # What waits, for the notice of a waiter ("the creation of an artifact" ...). + note = None, # A note left in the lock file for the waiters (who holds it and why). + ): + """ + Take the index lock for an operation of several steps that must not interleave with a full + reindex or with the same operation of another process (see `_lock_index`): the folder, the meta + and the record of a new artifact are made under it. Pass the lock to `add_to_index(index_lock=...)` + and release it in a `finally`. + + Returns: + dict: {'return': 0, 'lock': } or 'return' > 0 and 'error'. + """ + return self._lock_index(why, note) + + def _remove_index_leftovers( self, index_path, # The index folder (normalized). @@ -1098,8 +1122,12 @@ def _index( print (f'Repo file path: {repos_config_path}') - r = utils.files.safe_read_file(repos_config_path, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger) - if r['return']>0: return r + # Under the lock of repos.json, held until it is rewritten or left as it is: a plug or unplug of + # another process at the same moment is neither lost nor read half-written + r = utils.files.safe_read_file(repos_config_path, lock=True, keep_locked=True, retry_if_not_found=3, fail_on_error=self.fail_on_error, logger=self.logger) + if r['return']>0: return r + + repos_config_lock = r['file_lock'] paths_to_repos = {} original_paths_to_repos = r['data'] @@ -1145,7 +1173,10 @@ def _index( if to_update: # Do not sort keys - preserve order! - r = utils.files.safe_write_file(repos_config_path, paths_to_repos, atomic=True, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False) + r = utils.files.safe_write_file(repos_config_path, paths_to_repos, file_lock=repos_config_lock, atomic=True, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False) + if r['return']>0: return r + else: + r = utils.files.unlock_path(repos_config_path, repos_config_lock, fail_on_error=self.fail_on_error, logger=self.logger) if r['return']>0: return r if len(paths_to_repos) == 0: @@ -1516,15 +1547,12 @@ def _index( existing_category_index = {} index_file_lock = None - atomic_flag = False if clean: existing_category_index = category_index else: if os.path.isfile(index_artifact_file): - atomic_flag = True - r = utils.files.safe_read_file(index_artifact_file, lock=True, keep_locked=True, fail_on_error=self.fail_on_error, logger=self.logger) if r['return']>0: return r @@ -1570,8 +1598,9 @@ def _index( artifact_num += len(existing_category_index.get(self.KEY_INDEX_UIDS, {})) - r = utils.files.safe_write_file(index_artifact_file, existing_category_index, file_lock=index_file_lock, - atomic=atomic_flag, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False) + # Atomic for a new index file too: a reader never sees it half-written + r = utils.files.safe_write_file(index_artifact_file, existing_category_index, file_lock=index_file_lock, + atomic=True, fail_on_error=self.fail_on_error, logger=self.logger, sort_keys=False) if r['return']>0: return r diff --git a/cmeta/utils/files.py b/cmeta/utils/files.py index 3023db8..2e781de 100644 --- a/cmeta/utils/files.py +++ b/cmeta/utils/files.py @@ -7,6 +7,7 @@ """ import os +import sys import time import json import pickle @@ -50,14 +51,21 @@ LOCK_SOFT_REMOVE_SECONDS = 2.0 # how long a releasing soft lock retries the removal of its file (Windows: a reader may hold it) LOCK_IDENTITY_RETRIES = 50 # consecutive identity mismatches after which a path is locked with the soft lock LOCK_NOTE_MAX = 512 # bytes of the note a holder may leave in its lock file for the message of a waiter +LOCK_TIMEOUT_ENV = 'CMETA_LOCK_TIMEOUT' +LOCK_TIMEOUT = 30 # seconds a writer waits for the lock of a file (the environment variable changes it) +LOCK_NOTICE_SECONDS = 3 # a waiter says so on stderr after this long +READ_RETRY_SECONDS = 3 # how long a read without a lock is retried when the file cannot be read (a write in progress) +PICKLE_PROTOCOL = 4 # readable by every Python the engine supports, whatever the default of the Python that writes ERROR_CODE_FILE_NOT_FOUND = 16 -RETRY_DELAY = 0.1 +RETRY_DELAY = 0.1 # between the retries of a read; the longest delay between the retries of a replace RETRY_NOT_FOUND_FILE = 10 -RETRY_NOT_FOUND_INDEX_FILE = 2 -RETRY_REPLACE_FILE = 10 +RETRY_NOT_FOUND_INDEX_FILE = 3 # short retries of a cached file that is missing for a moment (a replace, the swap of a reindex) +RETRY_NOT_FOUND_DELAY = 0.025 RETRY_TIMESTAMP_FILE = 10 RETRY_DELETE_ATTEMPTS = 5 +_REPLACE_BUSY_ERRNOS = tuple(e for e in (getattr(errno, name, None) for name in ('EACCES', 'EPERM', 'EBUSY', 'ETXTBSY')) if e is not None) + ########################################################################################## def _get_lockfile_path( filepath: str, # Path to the file that needs locking. @@ -153,7 +161,9 @@ def is_path_within( # when left by another host) is renamed away and removed by the waiter that finds it. # In every mode the threads of one process are serialized on the path first (a threading.Lock per path), # which POSIX record locks and soft locks need (they are per process), and the busy wait polls with a -# delay that doubles from LOCK_POLL_START to LOCK_POLL_MAX until the timeout. +# delay that doubles from LOCK_POLL_START to LOCK_POLL_MAX until the timeout. The engine's own writers +# (_acquire_lock) wait LOCK_TIMEOUT seconds (CMETA_LOCK_TIMEOUT) and say so on stderr after +# LOCK_NOTICE_SECONDS, naming the holder when it left a note in the lock file (_acquire_with_notice). _LOCK_BUSY_ERRNOS = tuple(e for e in (getattr(errno, name, None) for name in ('EWOULDBLOCK', 'EAGAIN', 'EACCES', 'EDEADLK', 'EDEADLOCK')) if e is not None) _LOCK_UNSUPPORTED_ERRNOS = tuple(e for e in (getattr(errno, name, None) for name in ('ENOLCK', 'ENOSYS', 'EOPNOTSUPP', 'ENOTSUP', 'EINVAL', 'ENOTTY', 'EPERM', 'EBADF')) if e is not None) @@ -653,41 +663,123 @@ def _unlink_own_file(self): pass +########################################################################################## +def lock_timeout( + default: float = LOCK_TIMEOUT, # The seconds used when the environment variable is not set. + env_name: str = LOCK_TIMEOUT_ENV, # The environment variable that changes it. + logger = None, # Optional logger for the warning about a value that is not a number. +): + """ + The seconds a writer waits for a lock: `default` (LOCK_TIMEOUT, 30 s, for the lock of a file) + unless the environment variable `env_name` (CMETA_LOCK_TIMEOUT) holds a number of seconds; a value + that is not a number is ignored with a warning. + + Returns: + float: The seconds (0 or more). + """ + x = os.environ.get(env_name, '').strip() + if x != '': + try: + return max(float(x), 0.0) + except ValueError: + (logger if logger is not None else logging.getLogger('cmeta.utils.files')).warning( + f'{env_name}="{x}" is not a number of seconds - {default:g} is used') + return max(float(default), 0.0) + +########################################################################################## +def _acquire_with_notice( + lock, # The PathLock to take. + timeout: float, # Seconds to wait in all. + subject: str, # What is locked, for the messages ("the index in ", "''"). + what: str, # What waits ("the write of a record"), for the notice. + env_name: str, # The environment variable that changes the timeout, named in the messages. + note: str = None, # A note left in the lock file for the waiters (see PathLock.acquire). + logger = None, # Optional logger: the notice is a warning there too. +): + """ + Take `lock` within `timeout` seconds. A waiter that has not got it after LOCK_NOTICE_SECONDS says + so on stderr, naming the holder when it left a note in the lock file, and keeps waiting for the rest + of the time; when the lock stays held, the TimeoutError names the holder and the environment + variable that lengthens the wait. + + Returns: + PathLock: the lock, acquired. + + Raises: + TimeoutError: the lock stayed held for `timeout` seconds. + """ + timeout = max(float(timeout), 0.0) + first = min(float(LOCK_NOTICE_SECONDS), timeout) + + try: + return lock.acquire(timeout=first, note=note) + except TimeoutError as e: + holder = PathLock.read_note(lock.lock_file) + held_by = f' ({holder})' if holder else '' + if timeout <= first: + raise TimeoutError(f'{subject} stayed locked by another process{held_by} for {timeout:g} s ({e}) - set {env_name} to wait longer') + + message = (f'cMeta: {subject} is locked by another process{held_by} - ' + f'{what} continues when it is released (waiting up to {timeout:g} s; {env_name} changes it)') + print(message, file=sys.stderr, flush=True) + if logger is not None: + logger.warning(message) + + try: + return lock.acquire(timeout=max(timeout - first, 0.0), note=note) + except TimeoutError as e: + raise TimeoutError(f'{subject} stayed locked by another process{held_by} for {timeout:g} s ({e}) - set {env_name} to wait longer') + ########################################################################################## def _acquire_lock( filepath: str, # Path to the file to lock. - timeout: int = 3, # Maximum seconds to wait for lock acquisition. Default is 3. + timeout: float = None, # Seconds to wait for the lock; None: the engine's lock timeout (lock_timeout(): 30 s, CMETA_LOCK_TIMEOUT changes it). logger = None, # Optional logger for debug messages. + what: str = None, # What waits, named in the notice of a waiter ("the write of ''"). + note: str = None, # A note left in the lock file for the waiters (who holds it and why). ): """ Acquire the lock of a path (`.lock`, see PathLock) for process- and thread-safe - file operations. Blocks until the lock is acquired or the timeout expires. + file operations. Blocks until the lock is acquired or the timeout expires; a waiter says so on + stderr after LOCK_NOTICE_SECONDS (see _acquire_with_notice). Args: filepath (str): Path to the file to lock. - timeout (int): Maximum seconds to wait for lock acquisition. Default is 3. + timeout (float | None): Seconds to wait for the lock; None: the engine's lock timeout + (LOCK_TIMEOUT, 30 s; the environment variable CMETA_LOCK_TIMEOUT changes it). logger: Optional logger for debug messages. + what (str | None): What waits, named in the notice of a waiter. + note (str | None): A note left in the lock file for the waiters. Returns: PathLock: Acquired lock object that must be released later (`_release_lock`). Raises: - TimeoutError: If lock cannot be acquired within timeout period. + TimeoutError: If lock cannot be acquired within timeout period (the message names the holder + when it left a note), or for any other failure of the lock. """ lockfile = _get_lockfile_path(filepath) + if timeout is None: + timeout = lock_timeout(logger=logger) if logger is not None: logger.debug(f"utils.files._acquire_lock - attempting to create {lockfile} ...") file_lock = PathLock(lockfile, logger=logger) try: - file_lock.acquire(timeout=timeout) + _acquire_with_notice(file_lock, timeout, f"'{filepath}'", what if what is not None else f"the operation on '{filepath}'", + LOCK_TIMEOUT_ENV, note=note, logger=logger) + except TimeoutError: if logger is not None: - logger.debug(f"utils.files._acquire_lock - lock {lockfile} acquired!") - return file_lock + logger.debug(f"utils.files._acquire_lock - failed to lock {lockfile} in time ...") + raise except Exception as e: if logger is not None: logger.debug(f"utils.files._acquire_lock - failed to create {lockfile} ...") - raise TimeoutError(f"Could not acquire lock on '{filepath}' within {timeout} seconds: {str(e)}") + raise TimeoutError(f"Could not acquire lock on '{filepath}': {str(e)}") + + if logger is not None: + logger.debug(f"utils.files._acquire_lock - lock {lockfile} acquired!") + return file_lock ########################################################################################## def _check_lock( @@ -856,7 +948,7 @@ def safe_read_file( encoding: str = None, # Character encoding for text files. If None, auto-detected. lock: bool = False, # If True, acquires file lock before reading. keep_locked: bool = False, # If True with lock=True, keeps lock after read (returns in result). - timeout: int = 3, # Seconds to wait for lock acquisition. Default is 3. + timeout: float = None, # With lock=True, seconds to wait for the lock (None: the engine's lock timeout, 30 s, CMETA_LOCK_TIMEOUT changes it); without a lock, how long a read that fails is retried (None: READ_RETRY_SECONDS, 3 s; 0: one try). retry_if_not_found: int = 0, # Number of retry attempts if file not found. fail_on_error: bool = False, # If True, raises exception on error instead of returning error dict. logger = None, # Optional logger for debug messages. @@ -877,7 +969,8 @@ def safe_read_file( encoding: Character encoding for text files. If None, auto-detected. lock: If True, acquires file lock before reading. keep_locked: If True with lock=True, keeps lock after read (returns in result). - timeout: Seconds to wait for lock acquisition. Default is 3. + timeout: With a lock, seconds to wait for it (None: the engine's lock timeout, see lock_timeout); + without a lock, how long a read that fails is retried (None: READ_RETRY_SECONDS; 0: one try). retry_if_not_found: Number of retry attempts if file not found. fail_on_error: If True, raises exception on error instead of returning error dict. logger: Optional logger for debug messages. @@ -902,7 +995,7 @@ def safe_read_file( # to avoid cases, when file was deleted before writing if lock: try: - file_lock = _acquire_lock(filepath, timeout, logger) + file_lock = _acquire_lock(filepath, timeout, logger, what=f"the read of '{filepath}'") except Exception as e: return _error(None, 1, e, fail_on_error) @@ -937,7 +1030,10 @@ def safe_read_file( data = None if not lock: - if timeout == 0: + # Without a lock a read that fails (a write in progress) is retried for a while + retry = READ_RETRY_SECONDS if timeout is None else timeout + + if retry == 0: try: data = _read_file_data(filepath, encoding=encoding) except Exception as e: @@ -947,15 +1043,15 @@ def safe_read_file( # Retry reading with timeout when no lock start_time = time.time() last_error = None - - while time.time() - start_time < timeout: + + while time.time() - start_time < retry: try: data = _read_file_data(filepath, encoding=encoding) break # Success, exit retry loop - + except Exception as e: last_error = e - if time.time() - start_time < timeout: + if time.time() - start_time < retry: if logger is not None: logger.debug(f"utils.files.safe_read_file - retrying read file '{filepath}' due to error: {str(e)} ...") time.sleep(RETRY_DELAY) # Small delay before retry @@ -1057,7 +1153,7 @@ def write_file( else: yaml.safe_dump(data, f, sort_keys=sort_keys) elif file_format == "pickle": - pickle.dump(data, f) + pickle.dump(data, f, protocol=PICKLE_PROTOCOL) else: f.write(str(data)) @@ -1266,22 +1362,79 @@ def chunk(key): ########################################################################################## def _replace_file( - temp_path: str, # The file just written. - filepath: str, # The file it replaces. + temp_path: str, # The file just written (or the folder to rename). + filepath: str, # The file (or folder) it replaces. logger = None, # Optional logger for debug messages. + timeout: float = None, # Seconds the replace is retried while the target is busy (None: the engine's lock timeout). ): - """Replace a file atomically, with the retries that Windows needs after a lock.""" - for attempt in range(RETRY_REPLACE_FILE + 1): + """Replace a file atomically (`os.replace`; a folder is renamed the same way). The replace fails while + another process holds the target open (on Windows: a reader of an index file, a scanner, an antivirus) + or busy: it is retried with a delay that doubles from LOCK_POLL_START to RETRY_DELAY until `timeout` + seconds (the lock timeout, CMETA_LOCK_TIMEOUT) have passed. A target that cannot be written at all (a + read-only file, a folder without write permission) fails at once.""" + if timeout is None: + timeout = lock_timeout(logger=logger) + deadline = time.monotonic() + max(float(timeout), 0.0) + delay = LOCK_POLL_START + attempt = 0 + + while True: try: os.replace(temp_path, filepath) return - except Exception as e: - if attempt < RETRY_REPLACE_FILE: - if logger is not None: - logger.debug(f"utils.files._replace_file - retrying replace for '{temp_path}' -> '{filepath}' (attempt {attempt + 1}) due to error: {str(e)}") - time.sleep(RETRY_DELAY) - else: + except OSError as e: + if e.errno not in _REPLACE_BUSY_ERRNOS: + raise + attempt += 1 + remaining = deadline - time.monotonic() + if remaining <= 0 or not _replace_may_succeed_later(filepath): raise + if logger is not None: + logger.debug(f"utils.files._replace_file - retrying replace for '{temp_path}' -> '{filepath}' (attempt {attempt}) due to error: {str(e)}") + time.sleep(min(delay, remaining)) + delay = min(delay * 2, RETRY_DELAY) + +def _replace_may_succeed_later(filepath): + """False when a replace of `filepath` cannot succeed by waiting: the target is a read-only file, or its + folder cannot be written (on Windows os.access tells the read-only attribute of a file only).""" + try: + if os.path.isfile(filepath) and not os.access(filepath, os.W_OK): + return False + return os.access(os.path.dirname(filepath) or '.', os.W_OK) + except OSError: + return True + +########################################################################################## +def _install_file( + temp_path: str, # The file just written next to its target. + filepath: str, # The target. + logger = None, # Optional logger for debug messages. +): + """Put a file written next to its target in place with one atomic replace, so that a reader never sees + a half-written file: the mode of an existing target is kept (best effort); a target that is a symbolic + link is written through in place, so that the link stays (a replace would turn it into a plain file); a + target that cannot be written (read-only) is refused, as a write in place would be.""" + if os.path.islink(filepath): + with open(temp_path, 'rb') as src, open(filepath, 'wb') as dst: + shutil.copyfileobj(src, dst) + _remove_quietly(temp_path) + return + + if os.path.isfile(filepath): + if not os.access(filepath, os.W_OK): + raise PermissionError(errno.EACCES, 'Permission denied', filepath) + try: + shutil.copymode(filepath, temp_path) + except OSError: + pass + + _replace_file(temp_path, filepath, logger) + +def _remove_quietly(path): + try: + os.remove(path) + except OSError: + pass ########################################################################################## def _write_yaml_keeping_text( @@ -1336,7 +1489,7 @@ def _write_yaml_keeping_text( logger.debug(f"utils.files._write_yaml_keeping_text - the {method} of '{filepath}' does not load as expected: {loaded if not parsed else 'different data'}") if (parsed and loaded == data) or (method == 'dump' and parsed): - _replace_file(temp_path, filepath, logger) + _install_file(temp_path, filepath, logger) return {'return': 0, 'method': method, 'validated': parsed and loaded == data} finally: @@ -1352,9 +1505,9 @@ def _write_yaml_keeping_text( def safe_write_file( filepath: str, # Path where file should be written. data, # Data to write (dict/list for JSON/YAML, any object for pickle/text). - timeout: int = 3, # Seconds to wait for lock acquisition. Default is 3. + timeout: float = None, # Seconds to wait for the lock (None: the engine's lock timeout, 30 s, CMETA_LOCK_TIMEOUT changes it). file_lock = None, # Existing lock to use. If None, acquires new lock. - atomic: bool = False, # If True, writes to temp file then renames for atomicity. + atomic: bool = None, # If True, the file is written next to its target and put in place with one atomic replace, so that a reader never sees a half-written file and a crash leaves the previous one. None (the default): atomic for JSON, YAML and pickle, in place for text. encoding: str = None, # Character encoding for text files. If None, auto-detected. fail_on_error: bool = False, # If True, raises exception on error instead of returning error dict. logger = None, # Optional logger for debug messages. @@ -1381,9 +1534,12 @@ def safe_write_file( Args: filepath (str): Path where file should be written. data: Data to write (dict/list for JSON/YAML, any object for pickle/text). - timeout (int): Seconds to wait for lock acquisition. Default is 3. + timeout (float | None): Seconds to wait for the lock (None: the engine's lock timeout, see lock_timeout). file_lock: Existing lock to use. If None, acquires new lock. - atomic (bool): If True, writes to temp file then renames for atomicity. + atomic (bool | None): True: written next to the target and put in place with one atomic replace + (a reader never sees a half-written file; the mode of the target is kept; + a symbolic link is written through). None: atomic for JSON, YAML and + pickle, in place for text. False: in place. encoding (str | None): Character encoding for text files. If None, auto-detected. fail_on_error (bool): If True, raises exception on error instead of returning error dict. logger: Optional logger for debug messages. @@ -1407,12 +1563,17 @@ def safe_write_file( if file_lock is None: try: - file_lock = _acquire_lock(filepath, timeout, logger) + file_lock = _acquire_lock(filepath, timeout, logger, what=f"the write of '{filepath}'") except Exception as e: return _error(None, 1, e, fail_on_error) file_format = _detect_file_format(filepath) + # A structured file is written atomically unless told otherwise: a reader never sees a half-written + # file and a crash leaves the previous one (text is written in place, as write_file does) + if atomic is None: + atomic = file_format in ('json', 'yaml', 'pickle') + result = {'return': 0} release_error = None @@ -1439,18 +1600,23 @@ def safe_write_file( # A new YAML meta gets the keep style whether asked for it or for `preserve` keep_style = keep or (preserve and file_format == 'yaml') - r = write_file(temp_path, data, encoding=encoding, fail_on_error=fail_on_error, logger=logger, sort_keys=sort_keys, file_format=file_format, keep=keep_style) - if r['return']>0: return r + try: + r = write_file(temp_path, data, encoding=encoding, fail_on_error=fail_on_error, logger=logger, sort_keys=sort_keys, file_format=file_format, keep=keep_style) + except Exception: + if atomic: + _remove_quietly(temp_path) + raise + if r['return']>0: + if atomic: + _remove_quietly(temp_path) + return r if atomic: try: _check_lock(filepath, file_lock, logger) + _install_file(temp_path, filepath, logger) except Exception as e: - return _error(None, 1, e, fail_on_error) - - try: - _replace_file(temp_path, filepath, logger) - except Exception as e: + _remove_quietly(temp_path) return _error(None, 1, e, fail_on_error) finally: @@ -1469,7 +1635,7 @@ def safe_write_file( ########################################################################################## def safe_delete_directory( dirpath: str, # Full path to directory to delete. - timeout: int = 3, # Lock timeout in seconds. + timeout: float = None, # Seconds to wait for the lock (None: the engine's lock timeout, 30 s, CMETA_LOCK_TIMEOUT changes it). fail_on_error: bool = False, # Whether to raise exceptions or return error dict. logger = None, # Logger instance for debug messages. ): @@ -1506,7 +1672,7 @@ def safe_delete_directory( # Try to acquire lock try: - file_lock = _acquire_lock(dirpath, timeout, logger) + file_lock = _acquire_lock(dirpath, timeout, logger, what=f"the removal of '{dirpath}'") lock_acquired = True if logger is not None: logger.debug(f"utils.files.self_delete_directory - lock acquired for {dirpath}") @@ -1647,7 +1813,7 @@ def safe_delete_directory_if_empty( ########################################################################################## def lock_path( path: str, # Path to lock (file or directory). - timeout: int = 3, # Seconds to wait for lock acquisition. Default is 3. + timeout: float = None, # Seconds to wait for the lock (None: the engine's lock timeout, 30 s, CMETA_LOCK_TIMEOUT changes it). fail_on_error: bool = False, # If True, raises exception on error instead of returning error dict. logger = None, # Optional logger for debug messages. ): @@ -1671,7 +1837,7 @@ def lock_path( logger.debug(f"utils.files.lock_path - preparing to lock path {path} ...") try: - file_lock = _acquire_lock(path, timeout, logger) + file_lock = _acquire_lock(path, timeout, logger, what=f"the lock of '{path}'") return {'return': 0, 'file_lock': file_lock} except Exception as e: return _error(None, 1, e, fail_on_error) @@ -1710,24 +1876,30 @@ def unlock_path( return {'return':0} ########################################################################################## +_file_cache_guard = threading.Lock() # the cache dicts of safe_read_file_via_cache, for the threads of one process + def safe_read_file_via_cache( filepath: str, # Path to the file to read. cache: dict, # Dictionary to store cached data (modified in-place). - timeout: int = 10, # Lock timeout for file operations. + timeout: int = 10, # How long a read that fails (a write in progress) is retried, in seconds. fail_on_error: bool = False, # Whether to raise exceptions or return error dict. logger = None, # Optional logger for debug messages ): """ - Reads a file with caching based on file modification timestamp. - Automatically reloads if file has been modified since last cache. + Reads a file with caching keyed by the file's modification time in nanoseconds, size and inode + (an atomic replace changes the inode): the file is read again when any of them changed, which + catches two writes within one tick of a coarse file system too. A file read before that is + missing for a moment (being replaced on a mount whose rename is not atomic, or inside the swap + of a full reindex) is retried a few times before "not found". The cache dict is read and + updated under one lock for the threads of one process. The data handed out is the cached object + itself - do not change it (`Repos.find` copies what a caller may edit). - WARNING: This function is NOT thread-safe for async usage. The cache dictionary - can be corrupted by concurrent access, and it uses blocking I/O operations. + WARNING: This function uses blocking I/O operations. Args: filepath (str): Path to the file to read. cache (dict): Dictionary to store cached data (modified in-place). - timeout (int): Lock timeout for file operations. + timeout (int): How long a read that fails is retried, in seconds. fail_on_error (bool): Whether to raise exceptions or return error dict. logger: Optional logger for debug messages @@ -1738,55 +1910,64 @@ def safe_read_file_via_cache( Exception: Propagated runtime errors, if any. """ path = Path(filepath) - - if path.is_dir(): + + # Retry the stat RETRY_TIMESTAMP_FILE+1 times with small delay. A file this process has read before + # that is missing is given RETRY_NOT_FOUND_INDEX_FILE short retries: it is being replaced (on a mount + # whose rename is not atomic, 9p) or its folder is inside the swap of a full reindex; a file never + # seen is not retried (the index file of a category without artifacts) + not_found = 0 + for attempt in range(RETRY_TIMESTAMP_FILE + 1): + try: + st = path.stat() + break # Success, exit retry loop + except FileNotFoundError: + with _file_cache_guard: + seen = filepath in cache + if seen and not_found < RETRY_NOT_FOUND_INDEX_FILE: + not_found += 1 + time.sleep(RETRY_NOT_FOUND_DELAY) + continue + return _error(f"'{filepath}' does not exist", ERROR_CODE_FILE_NOT_FOUND, None, fail_on_error) + except Exception as e: + if attempt < RETRY_TIMESTAMP_FILE: # Don't delay after last attempt + if logger is not None: + logger.debug(f"utils.files.safe_read_file_via_cache - retrying stat for '{filepath}' (attempt {attempt + 1}) due to error: {str(e)}") + time.sleep(RETRY_DELAY) + else: + return _error(f"Could not get file timestamp: {str(e)}", 1, e, fail_on_error) + + if stat.S_ISDIR(st.st_mode): return _error(f"'{filepath}' is a directory; safe_read_file_via_cache only supports files.", 1, None, fail_on_error) - - for attempt in range(RETRY_NOT_FOUND_INDEX_FILE + 1): - if path.exists(): - break - return _error(f"'{filepath}' does not exist", ERROR_CODE_FILE_NOT_FOUND, None, fail_on_error) - - try: - # Retry getting file timestamp RETRY_TIMESTAMP_FILE+1 times with small delay - for attempt in range(RETRY_TIMESTAMP_FILE + 1): - try: - current_timestamp = path.stat().st_mtime - break # Success, exit retry loop - except Exception as e: - if attempt < RETRY_TIMESTAMP_FILE: # Don't delay after last attempt - if logger is not None: - logger.debug(f"utils.files.safe_read_file_via_cache - retrying getmtime for '{filepath}' (attempt {attempt + 1}) due to error: {str(e)}") - time.sleep(RETRY_DELAY) - else: - return _error(None, 1, e, fail_on_error) + stamp = (st.st_mtime_ns, st.st_size, st.st_ino) - except Exception as e: - return _error(f"Could not get file timestamp: {str(e)}", 1, e, fail_on_error) - - # Check if file is in cache and timestamp matches - if filepath in cache: - cached_entry = cache[filepath] - if cached_entry.get('timestamp') == current_timestamp: + # Check if file is in cache and its stamp matches + with _file_cache_guard: + cached_entry = cache.get(filepath) + + if cached_entry is not None: + if cached_entry.get('stamp') == stamp: return {'return': 0, 'data': cached_entry['data']} if logger is not None: logger.debug(f"utils.files.safe_read_file_via_cache - recaching changed index file {filepath} ...") - + # Cache miss or file changed - read the file if logger is not None: logger.debug(f"utils.files.safe_read_file_via_cache - reading index file {filepath} ...") result = safe_read_file(filepath, timeout=timeout, fail_on_error=fail_on_error, logger=logger) - + if result['return'] == 0: - # Update cache with new data and timestamp - cache[filepath] = { - 'data': result['data'], - 'timestamp': current_timestamp - } - + # Update cache with new data and stamp (a change of the file between the stat and the read is + # caught at the next call: the stamp stored is the older one) + with _file_cache_guard: + cache[filepath] = { + 'data': result['data'], + 'timestamp': st.st_mtime, + 'stamp': stamp + } + return result ########################################################################################## @@ -1794,7 +1975,7 @@ def safe_read_yaml_or_json( filepath: str, # Path to file (extension will be ignored/removed). lock: bool = False, # Whether to use file locking. keep_locked: bool = False, # Whether to keep lock after successful read. - timeout: int = 3, # Lock timeout. + timeout: float = None, # Lock timeout (None: the engine's lock timeout), or the read retry without a lock (see safe_read_file). fail_on_error: bool = False, # Whether to raise exceptions or return error dict. retry_if_not_found: int = 0, # Number of retries if file not found. logger = None, # Logger instance for debug messages. diff --git a/docs/async-and-concurrency.md b/docs/async-and-concurrency.md index 865c56c..2675fc8 100644 --- a/docs/async-and-concurrency.md +++ b/docs/async-and-concurrency.md @@ -211,22 +211,30 @@ r = utils.files.safe_write_file(index_file, index_data, - **File locks** are the engine's own `cmeta.utils.files.PathLock` (no library), using the sidecar file `.lock`, so they work cross-platform, *across - processes* and across the threads of one process. Default acquisition timeout - is **3 seconds**; exceeding it produces an error dict rather than a corrupt - file. + processes* and across the threads of one process. A writer waits up to + **30 seconds** for a busy lock (`CMETA_LOCK_TIMEOUT` changes it), prints a + notice on stderr after 3 seconds naming the holder when it left a note in the + lock file, and then gives up with an error dict that names the holder rather + than a corrupt file. - **`keep_locked=True`** hands the lock back with the data so the caller can modify and write without ever dropping it — this is what prevents lost updates. -- **`atomic=True`** writes to `.tmp` and then `os.replace()`s it into - position (retried up to 10 times), so a reader never observes a partially - written file, and a crash mid-write cannot truncate the original. +- **Atomic writes**: a JSON, YAML or pickle file is written to `.tmp` and + then `os.replace()`d into position (the default of `safe_write_file` since + 0.34.1; text is written in place), so a reader never observes a partially + written file, and a crash mid-write leaves the previous file and no + temporary. The mode of the target is kept, a symbolic link is written + through, a read-only target is refused, and the replace is retried while a + reader holds the target open (Windows) until the lock timeout. - Locks are released in a `finally:` block, and the lock file is removed by the process that releases it, so repositories never keep lock files. A crash leaves at most the lock file of the operation in flight; it blocks nobody (the OS lock dies with the process) and the next holder reuses and removes it. -The same mechanism protects `/index/*.pkl`, `repos.json`, and each -artifact's `_cmeta.yaml` / `_cmeta.json`. +The same mechanism protects `/index/*.pkl`, `repos.json` (also +while the incremental index of `plug` / `unplug` / `repo reindex` reads and +rewrites it), and each artifact's `_cmeta.yaml` / `_cmeta.json`, including the +one a `create` writes. ### How the lock works on each platform @@ -268,6 +276,16 @@ did; a write costs one more lock acquisition (microseconds on Linux, a fraction of a millisecond on Windows). Under the same lock a reindex removes the `index.tmp-*` / `index.old-*` leftovers of killed rebuilds. +### A create under the index lock + +`create` takes the index lock before it makes the folder, writes the meta and +adds the record, and releases it after (`Repos.lock_index`, +`add_to_index(index_lock=...)`): a full reindex never sees a half-made +artifact, several creates of one artifact at once end with one artifact and +"already exists" errors (the second finds the folder of the first), and a +create whose record cannot be written (the lock of its index file stayed held) +removes the folder it made, so nothing is left unindexed. + ### Measured behaviour Ten `cx note add` processes run simultaneously against one `` all @@ -282,10 +300,17 @@ files; `tests/benchmarks/benchmark_engine.py` measures the cost. ### What is *not* concurrency-safe -- **`safe_read_file_via_cache()`** — the in-memory cache dict it maintains is - explicitly documented as *not* thread-safe. It is the fast read path for the - index. This is precisely why `CMetaAsync` uses **processes, not threads**: - each worker gets its own `CMeta` and its own cache. +- **`safe_read_file_via_cache()`** keeps the in-memory cache of the index per + `CMeta` instance, keyed by the file's modification time in nanoseconds, size + and inode; an index file read before that is missing for a moment (a replace + on a mount whose rename is not atomic, the swap of a full reindex) is retried + a few times before "not found"; its dict is updated under a lock, and `find` hands out one-level + copies of the records it returns (the record, its `cmeta` and its `tags`), + so a caller may edit those. Anything deeper, and what the lower-level + `find_in_index` returns, is shared with the cache and must be treated as + read-only. `CMetaAsync` still uses + **processes, not threads**: each worker gets its own `CMeta` and its own + cache. - **Do not share one `CMeta` instance across threads.** Give each thread its own instance, or use `CMetaAsync` and let the process pool isolate them. - Long-running external work started by a task (builds, downloads) is *not* @@ -295,9 +320,10 @@ files; `tests/benchmarks/benchmark_engine.py` measures the cost. ### Practical notes -- **Lock contention shows up as an error, not corruption.** If you drive very - heavy parallel updates of the *same* artifact, expect occasional lock - timeouts; retry the call. +- **Lock contention shows up as a wait, then an error, not corruption.** A + writer waits up to `CMETA_LOCK_TIMEOUT` seconds (30) for the lock of a file + and says so on stderr after 3. If you drive very heavy parallel updates of + the *same* artifact, expect occasional lock timeouts; retry the call. - **Separate ``s remove sharing entirely.** For fully independent parallel jobs, give each one its own home (`--home=` or `CMETA_HOME`) — see [using-cmeta.md §7.4](using-cmeta.md#74-picking-cmeta_home-env-vars--per-project-collections). diff --git a/docs/using-cmeta.md b/docs/using-cmeta.md index 6ebe6ed..c542315 100644 --- a/docs/using-cmeta.md +++ b/docs/using-cmeta.md @@ -35,7 +35,9 @@ Environment variables worth knowing: `CMETA_HOME` (state directory, default under user home), `CMETA_DEBUG=1`, `CMETA_LOG=DEBUG|INFO`, `CMETA_LOG_FILE=`, `CMETA_VERBOSE=yes`, `CMETA_FAIL_ON_ERROR=yes`, `CMETA_INDEX_LOCK_TIMEOUT=` (how long a -write to the index waits for a running `cx --reindex`; 600). +write to the index waits for a running `cx --reindex`; 600), +`CMETA_LOCK_TIMEOUT=` (how long a writer waits for the lock of a file +held by another process; 30, with a notice after 3). --- diff --git a/tests/core_tests/test_utils_files_writes.py b/tests/core_tests/test_utils_files_writes.py new file mode 100644 index 0000000..bb47054 --- /dev/null +++ b/tests/core_tests/test_utils_files_writes.py @@ -0,0 +1,309 @@ +""" +Tests of the engine's writes and reads of shared files (cmeta.utils.files): the lock timeout and its +environment variable, the notice of a waiter, atomic writes by default for JSON, YAML and pickle (a reader +never sees a half-written file, a failed write leaves the previous file and no temporary), the mode of the +target kept, a symbolic link written through, a read-only target refused, the retries of a replace while a +reader holds the target, the reader cache keyed by time, size and inode, the cache under threads, the +pickle protocol of the index. + +cMeta author and developer: (C) 2025-2026 Grigori Fursin + +See the cMeta COPYRIGHT and LICENSE files in the project root for details. +""" + +import errno +import json +import os +import pickletools +import stat +import subprocess +import sys +import threading +import time + +import pytest + +from cmeta.utils import files + +POSIX = sys.platform != 'win32' +ROOT = POSIX and os.geteuid() == 0 + +HOLDER = r''' +import sys, time +from cmeta.utils import files +lock = files.PathLock(sys.argv[1]).acquire(timeout=60, note=sys.argv[2]) +print('LOCKED', flush=True) +time.sleep(float(sys.argv[3])) +lock.release() +print('RELEASED', flush=True) +''' + + +def hold(lock_file, note, seconds): + """Another process holds `lock_file` with `note` for `seconds`; returns once it holds it.""" + env = dict(os.environ) + env['PYTHONIOENCODING'] = 'utf-8' + p = subprocess.Popen([sys.executable, '-c', HOLDER, lock_file, note, str(seconds)], env=env, + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + line = p.stdout.readline().decode('utf-8', 'replace') + assert 'LOCKED' in line, line + return p + + +def stop(p): + p.kill() + p.communicate() + + +@pytest.fixture(autouse=True) +def no_lock_timeout_env(monkeypatch): + monkeypatch.delenv(files.LOCK_TIMEOUT_ENV, raising=False) + + +def test_the_lock_timeout_comes_from_the_environment(monkeypatch, caplog): + assert files.lock_timeout() == 30.0 + monkeypatch.setenv(files.LOCK_TIMEOUT_ENV, '7.5') + assert files.lock_timeout() == 7.5 + monkeypatch.setenv(files.LOCK_TIMEOUT_ENV, 'soon') + assert files.lock_timeout() == 30.0 + assert 'is not a number of seconds' in caplog.text + monkeypatch.delenv(files.LOCK_TIMEOUT_ENV) + assert files.lock_timeout(600, 'CMETA_OTHER_TIMEOUT_FOR_THIS_TEST') == 600.0 + + +def test_a_writer_waiting_for_a_file_lock_says_so_and_gives_up_after_the_limit(tmp_path, capfd): + target = str(tmp_path / 'shared.json') + assert files.safe_write_file(target, {'n': 0})['return'] == 0 + holder = hold(target + '.lock', 'a test holding the file', 60) + try: + t0 = time.time() + r = files.safe_write_file(target, {'n': 1}, timeout=5) + elapsed = time.time() - t0 + finally: + stop(holder) + + assert r['return'] > 0 + assert 'stayed locked by another process (a test holding the file)' in r['error'], r['error'] + assert files.LOCK_TIMEOUT_ENV in r['error'], r['error'] + assert 4.5 <= elapsed < 60, elapsed + err = capfd.readouterr().err + assert 'is locked by another process (a test holding the file)' in err and 'the write of' in err, err + assert json.load(open(target)) == {'n': 0} + assert not os.path.exists(target + '.tmp') + + +def test_structured_files_are_written_atomically_by_default(tmp_path): + for name, data in (('a.json', {'a': 1}), ('a.yaml', {'a': 1}), ('a.pkl', {'a': 1})): + target = str(tmp_path / name) + r = files.safe_write_file(target, data) + assert r['return'] == 0, r + assert files.safe_read_file(target)['data'] == data + assert not os.path.exists(target + '.tmp') + assert not os.path.exists(target + '.lock') + + # A write that fails leaves the previous file and no temporary + target = str(tmp_path / 'a.json') + r = files.safe_write_file(target, {'a': object()}) + assert r['return'] > 0 + assert files.safe_read_file(target)['data'] == {'a': 1} + assert not os.path.exists(target + '.tmp') + + # Text is written in place, without a temporary + target = str(tmp_path / 'a.txt') + assert files.safe_write_file(target, 'hello')['return'] == 0 + assert open(target).read() == 'hello' + assert not os.path.exists(target + '.tmp') + + +def test_the_mode_of_the_target_is_kept_and_a_read_only_target_is_refused(tmp_path): + target = str(tmp_path / 'm.json') + assert files.safe_write_file(target, {'a': 1})['return'] == 0 + expected = {'a': 1} + + if POSIX: + os.chmod(target, 0o640) + if stat.S_IMODE(os.stat(target).st_mode) != 0o640: + pytest.skip('the file system keeps no mode (a 9p or FAT mount)') + assert files.safe_write_file(target, {'a': 2})['return'] == 0 + assert stat.S_IMODE(os.stat(target).st_mode) == 0o640 + expected = {'a': 2} + assert files.safe_read_file(target)['data'] == expected + + if ROOT: + return # root writes read-only files, as it always did + + os.chmod(target, stat.S_IREAD) + try: + if os.access(target, os.W_OK): + pytest.skip('the file system keeps no read-only mode (a 9p or FAT mount)') + r = files.safe_write_file(target, {'a': 3}) + assert r['return'] > 0, r + assert files.safe_read_file(target)['data'] == expected + assert not os.path.exists(target + '.tmp') + finally: + os.chmod(target, stat.S_IWRITE | stat.S_IREAD) + + +@pytest.mark.skipif(not POSIX, reason='symbolic links need a privilege on Windows') +def test_a_symbolic_link_is_written_through(tmp_path): + real = str(tmp_path / 'real.json') + link = str(tmp_path / 'link.json') + assert files.safe_write_file(real, {'a': 1})['return'] == 0 + os.symlink(real, link) + assert files.safe_write_file(link, {'a': 2})['return'] == 0 + assert os.path.islink(link) + assert json.load(open(real)) == {'a': 2} + assert not os.path.exists(link + '.tmp') + + +def test_a_replace_waits_for_a_target_that_is_busy_and_gives_up_after_the_lock_timeout(tmp_path, monkeypatch): + target = str(tmp_path / 'held.json') + temp = target + '.tmp' + with open(target, 'w') as f: + f.write('{"a": 0}') + with open(temp, 'w') as f: + f.write('{"a": 1}') + + if not POSIX: + # Windows: a reader holding the target open makes the replace fail until it closes it + def read_for_a_while(seconds): + with open(target): + time.sleep(seconds) + t = threading.Thread(target=read_for_a_while, args=(2.0,)) + t.start() + time.sleep(0.2) + t0 = time.time() + files._replace_file(temp, target) + elapsed = time.time() - t0 + t.join() + else: + # POSIX: os.replace itself never waits for a reader; a busy target is shown with a hook + real = os.replace + until = time.monotonic() + 1.5 + + def busy(a, b): + if time.monotonic() < until: + raise PermissionError(errno.EACCES, 'busy for the test') + return real(a, b) + + monkeypatch.setattr(os, 'replace', busy) + t0 = time.time() + files._replace_file(temp, target) + elapsed = time.time() - t0 + monkeypatch.setattr(os, 'replace', real) + + assert 1.0 < elapsed < 20, elapsed + assert json.load(open(target)) == {'a': 1} + + # The wait is bounded by the lock timeout + with open(temp, 'w') as f: + f.write('{"a": 2}') + monkeypatch.setenv(files.LOCK_TIMEOUT_ENV, '1') + + if not POSIX: + t = threading.Thread(target=read_for_a_while, args=(4.0,)) + t.start() + time.sleep(0.2) + t0 = time.time() + with pytest.raises(PermissionError): + files._replace_file(temp, target) + elapsed = time.time() - t0 + t.join() + else: + def always_busy(a, b): + raise PermissionError(errno.EACCES, 'busy for the test') + monkeypatch.setattr(os, 'replace', always_busy) + t0 = time.time() + with pytest.raises(PermissionError): + files._replace_file(temp, target) + elapsed = time.time() - t0 + monkeypatch.setattr(os, 'replace', real) + + assert 0.9 <= elapsed < 10, elapsed + assert json.load(open(target)) == {'a': 1} + os.remove(temp) + + +def test_the_cache_sees_a_change_that_keeps_the_timestamp(tmp_path): + target = str(tmp_path / 'c.json') + cache = {} + assert files.safe_write_file(target, {'a': 1})['return'] == 0 + st = os.stat(target) + assert files.safe_read_file_via_cache(target, cache)['data'] == {'a': 1} + assert cache[target]['stamp'] == (st.st_mtime_ns, st.st_size, st.st_ino) + + # A change of size within the same timestamp + assert files.safe_write_file(target, {'a': 22})['return'] == 0 + os.utime(target, ns=(st.st_atime_ns, st.st_mtime_ns)) + assert files.safe_read_file_via_cache(target, cache)['data'] == {'a': 22} + + # A change of the same size within the same timestamp: the replaced file has another inode (where + # the file system reports inodes) + assert files.safe_write_file(target, {'a': 33})['return'] == 0 + os.utime(target, ns=(st.st_atime_ns, st.st_mtime_ns)) + if os.stat(target).st_ino: + assert files.safe_read_file_via_cache(target, cache)['data'] == {'a': 33} + + # Unchanged: served from the cache (the same object) + first = files.safe_read_file_via_cache(target, cache)['data'] + assert files.safe_read_file_via_cache(target, cache)['data'] is first + + # A missing file never seen: the not-found code at once; a file read before that is missing is + # retried for a moment (a replace in progress, the swap of a reindex) and found again + t0 = time.time() + r = files.safe_read_file_via_cache(str(tmp_path / 'absent.json'), cache) + assert r['return'] == files.ERROR_CODE_FILE_NOT_FOUND + assert time.time() - t0 < 0.5 + r = files.safe_read_file_via_cache(str(tmp_path), cache) + assert r['return'] == 1 and 'is a directory' in r['error'] + + gone = target + '.away' + os.replace(target, gone) + threading.Timer(0.03, lambda: os.replace(gone, target)).start() + r = files.safe_read_file_via_cache(target, cache) + assert r['return'] == 0 and r['data'] == {'a': 33}, r + os.replace(target, gone) + t0 = time.time() + r = files.safe_read_file_via_cache(target, cache) + assert r['return'] == files.ERROR_CODE_FILE_NOT_FOUND + assert 0.05 < time.time() - t0 < 2, time.time() - t0 + os.replace(gone, target) + + +def test_the_cache_is_safe_for_threads(tmp_path): + target = str(tmp_path / 't.json') + assert files.safe_write_file(target, {'n': 0})['return'] == 0 + cache = {} + errors = [] + + def reader(): + try: + for i in range(100): + r = files.safe_read_file_via_cache(target, cache) + assert r['return'] == 0 and 'n' in r['data'], r + time.sleep(0.001) + except Exception as e: + errors.append(e) + + def writer(): + for i in range(1, 11): + r = files.safe_write_file(target, {'n': i}) + assert r['return'] == 0, r + time.sleep(0.01) + + threads = [threading.Thread(target=reader) for _ in range(4)] + [threading.Thread(target=writer)] + for t in threads: + t.start() + for t in threads: + t.join() + assert errors == [] + assert files.safe_read_file_via_cache(target, cache)['data'] == {'n': 10} + + +def test_a_pickle_is_written_with_protocol_4(tmp_path): + target = str(tmp_path / 'p.pkl') + assert files.safe_write_file(target, {'a': [1, 2]})['return'] == 0 + with open(target, 'rb') as f: + ops = list(pickletools.genops(f)) + assert ops[0][0].name == 'PROTO' and ops[0][1] == files.PICKLE_PROTOCOL == 4 + assert files.safe_read_file(target)['data'] == {'a': [1, 2]} diff --git a/tests/internal_repo_tests/test_integration_create_guard.py b/tests/internal_repo_tests/test_integration_create_guard.py new file mode 100644 index 0000000..5ec8183 --- /dev/null +++ b/tests/internal_repo_tests/test_integration_create_guard.py @@ -0,0 +1,186 @@ +""" +Integration tests of a create under the index lock and of the writes around it: creates of one artifact +at once make one artifact, a create whose record cannot be written leaves nothing behind (and says who +holds the lock), the index of repositories takes the lock of repos.json, a find result can be edited +without changing the cached index of the process. + +Each test runs against a fresh in tmp_path so it does not touch the user's real cMeta +state. + +cMeta author and developer: (C) 2025-2026 Grigori Fursin + +See the cMeta COPYRIGHT and LICENSE files in the project root for details. +""" + +import os +import pickle +import subprocess +import sys +import time + +import pytest + +from cmeta import CMeta +from cmeta import repos as cmeta_repos + + +@pytest.fixture() +def cm(tmp_path, monkeypatch): + for var in ('CMETA_HOME', 'CMETA_HOME2', 'VIRTUAL_ENV', 'CONDA_PREFIX', + 'CMETA_DEBUG', 'CMETA_VERBOSE', 'CMETA_FAIL_ON_ERROR', 'CMETA_INDEX_LOCK_TIMEOUT', 'CMETA_LOCK_TIMEOUT'): + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv('CMETA_HOME', str(tmp_path)) + monkeypatch.setattr(cmeta_repos, '_migrated_notices', set()) + return CMeta(home=str(tmp_path)) + + +def make(cm, alias, **meta): + r = cm.access({'category': 'log', 'command': 'create', 'arg1': f'local:{alias}', 'meta': meta, 'con': False}) + assert r['return'] == 0, r.get('error') + return r['meta']['artifact'] + + +def env_for(home, **extra): + env = {k: v for k, v in os.environ.items() if not k.startswith('CMETA_') and k not in ('VIRTUAL_ENV', 'CONDA_PREFIX')} + env['CMETA_HOME'] = str(home) + env['PYTHONIOENCODING'] = 'utf-8' + env.update(extra) + return env + + +def start(args, home, **extra): + """A cMeta command line (cx ) in another process, with a null input; its output is read by finish().""" + return subprocess.Popen([sys.executable, '-m', 'cmeta'] + list(args), env=env_for(home, **extra), + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + + +def finish(proc, timeout=300): + out, _ = proc.communicate(timeout=timeout) + return proc.returncode, out.decode('utf-8', 'replace') + + +HOLDER = r''' +import sys, time +from cmeta.utils import files +lock = files.PathLock(sys.argv[1]).acquire(timeout=60, note=sys.argv[2]) +print('LOCKED', flush=True) +time.sleep(float(sys.argv[3])) +lock.release() +''' + + +def hold(lock_file, note, seconds=60): + """Another process holds `lock_file` with `note` for `seconds`; returns once it holds it.""" + env = dict(os.environ) + env['PYTHONIOENCODING'] = 'utf-8' + p = subprocess.Popen([sys.executable, '-c', HOLDER, str(lock_file), note, str(seconds)], env=env, + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + line = p.stdout.readline().decode('utf-8', 'replace') + assert 'LOCKED' in line, line + return p + + +def stop(p): + p.kill() + p.communicate() + + +def index_aliases(tmp_path, category='log'): + """(the lowercase aliases of the index, {uid: path}) of a category as the pickle on disk holds them.""" + with open(tmp_path / 'index' / f'{category}.pkl', 'rb') as f: + data = pickle.load(f) + return set(data['lowercase_aliases']), {uid: rec['path'] for uid, rec in data['uids'].items()} + + +def folders(home, category='log'): + base = home / 'repos' / 'local' / category + return {p.name.lower() for p in base.iterdir() if p.is_dir()} if base.is_dir() else set() + + +def test_creates_of_one_alias_at_once_make_one_artifact(cm, tmp_path): + make(cm, 'present') + + procs = [start(['log', 'create', 'twin'], tmp_path) for _ in range(6)] + results = [finish(p) for p in procs] + + codes = [rc for rc, _ in results] + assert codes.count(0) == 1, results + for rc, out in results: + if rc != 0: + assert rc == 8 and 'already exist' in out, (rc, out) + + aliases, paths = index_aliases(tmp_path) + assert folders(tmp_path) == {'present', 'twin'} + assert aliases == {'present', 'twin'} + assert len([uid for uid, path in paths.items() if os.path.basename(path).lower() == 'twin']) == 1 + assert all(os.path.isdir(p) for p in paths.values()) + assert not (tmp_path / 'index.lock').exists() + + +def test_a_create_whose_record_cannot_be_written_leaves_nothing_behind(cm, tmp_path): + make(cm, 'present') + index_file = tmp_path / 'index' / 'log.pkl' + + holder = hold(str(index_file) + '.lock', 'a test holding the index file') + try: + t0 = time.time() + rc, out = finish(start(['log', 'create', 'late'], tmp_path, CMETA_LOCK_TIMEOUT='5'), timeout=120) + elapsed = time.time() - t0 + finally: + stop(holder) + + assert rc != 0, out + assert 'is locked by another process (a test holding the index file)' in out and 'the read of' in out, out + assert 'stayed locked by another process (a test holding the index file)' in out and 'CMETA_LOCK_TIMEOUT' in out, out + assert 4.5 <= elapsed < 60, elapsed + + # Nothing of the failed create is left: no folder, no record, no index lock + assert folders(tmp_path) == {'present'} + assert index_aliases(tmp_path)[0] == {'present'} + assert not (tmp_path / 'index.lock').exists() + + +def test_the_index_of_repositories_takes_the_lock_of_repos_json(cm, tmp_path): + make(cm, 'present') + + holder = hold(str(tmp_path / 'repos.json') + '.lock', 'a test holding repos.json') + try: + rc, out = finish(start(['repo', 'reindex', 'local'], tmp_path, CMETA_LOCK_TIMEOUT='2'), timeout=120) + finally: + stop(holder) + + assert rc != 0, out + assert 'repos.json' in out and 'stayed locked by another process (a test holding repos.json)' in out, out + assert index_aliases(tmp_path)[0] == {'present'} + assert not (tmp_path / 'index.lock').exists() + + # Released: the same command works + rc, out = finish(start(['repo', 'reindex', 'local'], tmp_path), timeout=120) + assert rc == 0, out + assert index_aliases(tmp_path)[0] == {'present'} + + +def test_a_find_result_can_be_edited_without_changing_the_index_of_the_process(cm, tmp_path): + make(cm, 'tagged', tags=['one']) + + r = cm.access({'category': 'log', 'command': 'find', 'arg1': 'tagged', 'con': False}) + assert r['return'] == 0, r.get('error') + artifact = r['artifacts'][0] + artifact['cmeta']['tags'].append('added-by-the-caller') + artifact['cmeta']['new_key'] = 1 + artifact['cmeta_ref_parts'] = {'replaced': True} + artifact['path'] = 'elsewhere' + + # The same through a tag filter (the records are copied after the filters, for what is returned) + r = cm.access({'category': 'log', 'command': 'find', 'arg1': 'tagged', 'tags': 'one', 'con': False}) + assert r['return'] == 0, r.get('error') + assert r['artifacts'][0]['cmeta']['tags'] == ['one'] + r['artifacts'][0]['cmeta']['tags'].append('added-through-the-filter') + + r = cm.access({'category': 'log', 'command': 'find', 'arg1': 'tagged', 'con': False}) + assert r['return'] == 0, r.get('error') + again = r['artifacts'][0] + assert again['cmeta']['tags'] == ['one'] + assert 'new_key' not in again['cmeta'] + assert again['cmeta_ref_parts']['artifact_alias'] == 'tagged' + assert again['path'] != 'elsewhere' diff --git a/tests/internal_repo_tests/test_integration_index_guard.py b/tests/internal_repo_tests/test_integration_index_guard.py index fb16afe..c48e679 100644 --- a/tests/internal_repo_tests/test_integration_index_guard.py +++ b/tests/internal_repo_tests/test_integration_index_guard.py @@ -31,7 +31,7 @@ @pytest.fixture() def cm(tmp_path, monkeypatch): for var in ('CMETA_HOME', 'CMETA_HOME2', 'VIRTUAL_ENV', 'CONDA_PREFIX', - 'CMETA_DEBUG', 'CMETA_VERBOSE', 'CMETA_FAIL_ON_ERROR', 'CMETA_INDEX_LOCK_TIMEOUT'): + 'CMETA_DEBUG', 'CMETA_VERBOSE', 'CMETA_FAIL_ON_ERROR', 'CMETA_INDEX_LOCK_TIMEOUT', 'CMETA_LOCK_TIMEOUT'): monkeypatch.delenv(var, raising=False) monkeypatch.setenv('CMETA_HOME', str(tmp_path)) monkeypatch.setattr(cmeta_repos, '_migrated_notices', set()) @@ -131,9 +131,11 @@ def test_writes_during_a_full_reindex_wait_for_it_and_land_in_the_new_index(cm, } every = [p for group in writers.values() for p in group] - # While the reindex holds the lock none of the writers gets through + # While the reindex holds the lock none of the writers gets through, and a create waits BEFORE it + # writes its folder: nothing unindexed appears on disk meanwhile time.sleep(2.0) assert all(p.poll() is None for p in every), 'a writer finished while the reindex held the index lock' + assert not any(name.startswith('born-') for name in folders(tmp_path)), folders(tmp_path) go.touch() rc, out = finish(reindex) From ce11b6cbcb20e921dcdf618a05fd718e4877b1fb Mon Sep 17 00:00:00 2001 From: Grigori Fursin Date: Fri, 9 Oct 2026 15:00:28 +0200 Subject: [PATCH 4/4] cMeta v0.34.1 Co-Authored-By: Claude Fable 5.1 Signed-off-by: Grigori Fursin --- CHANGELOG.md | 2 +- CITATION.cff | 2 +- cmeta/internal-repo/_cmr.yaml | 2 +- cmeta/version.py | 2 +- cmeta/versions.json | 4 ++++ docs/en/api/index.rst | 4 ++-- docs/en/index.rst | 4 ++-- docs/history.md | 2 +- docs/installation.md | 4 ++-- 9 files changed, 15 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d78c386..84d638e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,7 @@ All notable changes to cMeta are documented here, newest first. -## 0.34.0.1 (in development) +## 0.34.1 - **The index lock: a full reindex no longer loses the writes made while it runs, and two rebuilds no longer race for the swap.** A full `cx --reindex` builds the new index in a temporary folder and swaps it in when it is complete; a record written to the live index meanwhile by another process (a diff --git a/CITATION.cff b/CITATION.cff index 6225fc3..2fce0a9 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -13,7 +13,7 @@ authors: website: "https://cTuning.ai/@gfursin" affiliation: "cTuning Labs" orcid: "https://orcid.org/0000-0001-7719-1624" -version: "0.34.0" +version: "0.34.1" date-released: "2026-10-09" license: Apache-2.0 repository-code: "https://github.com/cTuningLabs/cmeta" diff --git a/cmeta/internal-repo/_cmr.yaml b/cmeta/internal-repo/_cmr.yaml index f91330f..604e313 100644 --- a/cmeta/internal-repo/_cmr.yaml +++ b/cmeta/internal-repo/_cmr.yaml @@ -3,5 +3,5 @@ authors: "[Grigori Fursin](https://cTuning.ai/@gfursin)" category: repo,f4f792ab40c7498f copyright: "2025-2026 Grigori Fursin and cTuning Labs. See the cMeta COPYRIGHT and LICENSE files for details." permanent: True -version: '0.34.0' +version: '0.34.1' version_date: '20261009' diff --git a/cmeta/version.py b/cmeta/version.py index 9289284..376b70e 100644 --- a/cmeta/version.py +++ b/cmeta/version.py @@ -6,5 +6,5 @@ See the cMeta COPYRIGHT and LICENSE files in the project root for details. """ -__version__ = "0.34.0.1" +__version__ = "0.34.1" __release_date__ = "2026-10-09" diff --git a/cmeta/versions.json b/cmeta/versions.json index 09868e6..d66c15d 100644 --- a/cmeta/versions.json +++ b/cmeta/versions.json @@ -93,6 +93,10 @@ { "version": "0.34.0", "version_date": "20261009" + }, + { + "version": "0.34.1", + "version_date": "20261009" } ] } diff --git a/docs/en/api/index.rst b/docs/en/api/index.rst index c409e27..769ab6a 100644 --- a/docs/en/api/index.rst +++ b/docs/en/api/index.rst @@ -1,7 +1,7 @@ -API Reference (v0.34.0) +API Reference (v0.34.1) ======================= -This section contains the complete API reference for cMeta v0.34.0. +This section contains the complete API reference for cMeta v0.34.1. .. toctree:: :maxdepth: 2 diff --git a/docs/en/index.rst b/docs/en/index.rst index 3811c7f..a42d28d 100644 --- a/docs/en/index.rst +++ b/docs/en/index.rst @@ -1,11 +1,11 @@ cMeta Documentation =================== -Welcome to cMeta v0.34.0 documentation! +Welcome to cMeta v0.34.1 documentation! .. toctree:: :maxdepth: 2 - :caption: cMeta v0.34.0 Contents: + :caption: cMeta v0.34.1 Contents: home diff --git a/docs/history.md b/docs/history.md index 1628e9a..3201c58 100644 --- a/docs/history.md +++ b/docs/history.md @@ -82,7 +82,7 @@ produces APA and BibTeX automatically. The BibTeX entry: author = {Fursin, Grigori}, title = {{cMeta (Common Meta Framework)}}, year = {2026}, - version = {0.34.0}, + version = {0.34.1}, license = {Apache-2.0}, url = {https://github.com/cTuningLabs/cmeta}, note = {cTuning Labs} diff --git a/docs/installation.md b/docs/installation.md index 44e62ea..b579b88 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -157,8 +157,8 @@ repositories) into these commands. **A specific version**, to pin or to go back: ```bash -uv tool install --force "cmeta[server]==0.34.0" -pip install "cmeta==0.34.0" +uv tool install --force "cmeta[server]==0.34.1" +pip install "cmeta==0.34.1" ``` **Moving an install from git to PyPI** (or back) is a reinstall from the other