diff --git a/src/did/did_matlab_python_bridge_file.yaml b/src/did/did_matlab_python_bridge_file.yaml index 3a98582..3928c14 100644 --- a/src/did/did_matlab_python_bridge_file.yaml +++ b/src/did/did_matlab_python_bridge_file.yaml @@ -300,7 +300,7 @@ classes: - name: binaryTable type: class matlab_path: "+did/+file/binaryTable.m" - matlab_last_sync_hash: "4a0f9a5" + matlab_last_sync_hash: "bd369da" status: ported python_path: "did/file.py" python_class: "BinaryTable" @@ -399,9 +399,28 @@ classes: key and the matching release is a no-op, so the outermost caller owns the lock. + PYTHON-ONLY ADDITION (2026-10-23): a per-table threading.Lock, held + from get_lock to the outermost release_lock, and has_lock counts as + held only for the thread that took it. The cache from get_cache() + is one object per process, and has_lock was one flag on it: while + one thread held the lock, every other thread took the "already + held" path and rewrote the index unlocked (an NDI viewer adding + from a pool of download threads lost index rows). MATLAB runs one + thread and has nothing to follow; reset_lock_state likewise only + clears a hold of the calling thread's. + + 2026-10-23: the lock now expires after 20 s with 30 s of waiting, + the arguments MATLAB's getLock passes (checkout_lock_file(..., + 30,1,20)). Python used checkout_lock_file's defaults, a one-hour + expiry, so a holder killed mid-update (an NDI viewer quitting + while a daemon thread added to the cache) locked every other + process out of the cache for an hour. + - name: releaseLock python_name: release_lock - decision_log: "Exact match." + decision_log: > + Exact match, plus releasing the per-thread hold get_lock takes + (Python only; see getLock). - name: lockFileName python_name: lock_file_name @@ -496,6 +515,9 @@ classes: two agree on where a missing value belongs and not merely on whether it is present. + 2026-10-23: the unsorted path reads the column once instead of + reopening the file per row. Same result; the same change as DID-matlab#221. + decision_log: > SYNCHRONIZED 2026-08-29. Previously a partial port: read_row returned raw bytes and insertRow, deleteRow, writeEntry, writeTable, findRow and @@ -508,7 +530,7 @@ classes: - name: fileCache type: class matlab_path: "+did/+file/fileCache.m" - matlab_last_sync_hash: "4a0f9a5" + matlab_last_sync_hash: "3041f77" status: ported python_path: "did/file.py" python_class: "FileCache" @@ -654,7 +676,8 @@ classes: Ported 2026-08-29. Both move the file in by default and copy on request (MATLAB 'copy',true; Python copy=True), both refuse a name whose length is not fileNameCharacters, and both refuse a name - already in the cache. + already in the cache. 2026-10-23: its lookup is a binary search, + as isFile; DID-matlab#221. - name: removeFile python_name: remove_file @@ -670,7 +693,10 @@ classes: - name: isFile python_name: is_file - decision_log: "Ported 2026-08-29. Both search the index by name." + decision_log: > + Ported 2026-08-29. Both search the index by name. 2026-10-23: a + binary search (sorted=True), since the index is kept sorted and a + scan grew with the cache; the same change as DID-matlab#221. - name: fileList python_name: file_list @@ -700,7 +726,8 @@ classes: python_name: touch decision_log: > Ported 2026-08-29. Both update the last-access time and report - whether the name was in the index. + whether the name was in the index. 2026-10-23: binary search, as + isFile; DID-matlab#221. - name: infoFileName python_name: _info_file_name @@ -751,7 +778,7 @@ functions: - name: checkout_lock_file type: function matlab_path: "+did/+file/checkout_lock_file.m" - matlab_last_sync_hash: "9010a73" + matlab_last_sync_hash: "bd369da" status: ported python_path: "did/file.py" python_name: "checkout_lock_file" @@ -768,6 +795,15 @@ functions: see the parse_lock_expiration entry below. Both write the expiry as ISO 8601 and both read either format. + 2026-10-23, with DID-matlab#221: an optional + max_poll_interval (MATLAB: a fifth argument, maxPollInterval). Given, + a waiter looks again after 5 ms and backs off to at most that + interval, within check_loops seconds in all; omitted, it looks once a + second as before. BinaryTable.get_lock passes 0.1: a cache-index + update holds the lock for milliseconds, so a one-second check cost + each waiter a second per collision. Callers holding locks for long + operations keep the old cadence. + - name: lock_file_key type: function matlab_path: "+did/+file/lock_file_key.m" diff --git a/src/did/file.py b/src/did/file.py index f84e9df..475cafa 100644 --- a/src/did/file.py +++ b/src/did/file.py @@ -4,6 +4,7 @@ import re import shutil import struct +import threading import time import uuid from datetime import datetime, timedelta, timezone @@ -247,9 +248,18 @@ def parse_lock_expiration(text): ) -def checkout_lock_file(filename, check_loops=30, throw_error=True, expiration=3600): +def checkout_lock_file( + filename, check_loops=30, throw_error=True, expiration=3600, max_poll_interval=None +): """Try to establish control of the lock file named ``filename``. + While another holder has it, wait up to ``check_loops`` seconds, + looking again once a second. A caller whose lock is only ever held + briefly can pass ``max_poll_interval`` (seconds): the first look again + is then after 5 ms, backing off to at most that interval, so a lock + held for milliseconds is handed on almost at once. The file cache's + index lock passes 0.1; other callers keep the one-second checks. + ``filename`` is the lock file itself, not the file being protected -- the same contract as MATLAB's did.file.checkout_lock_file, whose own docstring shows a caller passing "myfile.txt-lock". This used to append @@ -260,7 +270,15 @@ def checkout_lock_file(filename, check_loops=30, throw_error=True, expiration=36 key = f"{_utcnow().isoformat()}_{uuid.uuid4()}" lock_filename = filename - for _ in range(check_loops): + deadline = time.monotonic() + check_loops + if max_poll_interval is None: + delay = max_delay = 1.0 + else: + max_delay = max(float(max_poll_interval), 0.001) + delay = min(0.005, max_delay) + first = True + while first or time.monotonic() < deadline: + first = False try: # Not a context manager: the handle is returned to the caller, # which holds the lock until release_lock_file(). @@ -288,7 +306,8 @@ def checkout_lock_file(filename, check_loops=30, throw_error=True, expiration=36 except (OSError, ValueError): # Could not read lock file or parse time, wait and retry pass - time.sleep(1) + time.sleep(delay) + delay = min(delay * 2, max_delay) if throw_error: raise OSError(f"Unable to obtain lock with file {filename}.") @@ -370,6 +389,13 @@ def __init__(self, f, record_type, record_size, elements_per_column, header_size self.elements_per_column = [int(n) for n in elements_per_column] self.header_size = int(header_size) self.has_lock = False + # The lock file serialises processes; this serialises the threads of + # one process. has_lock alone cannot: it is one flag on a shared + # object (FileCache is a process-wide singleton), so while one + # thread held the lock every other thread saw has_lock True, took + # the "already held" path, and changed the table unlocked. + self._thread_lock = threading.Lock() + self._lock_owner = None self.file.set_properties(machineformat="l") # always little-endian if not self.file.fullpathfilename: @@ -548,16 +574,44 @@ def get_lock(self): held, which is what makes ``release_lock`` a no-op for the nested call and leaves the outermost caller owning the lock. """ - if not self.has_lock: - lock_fid, key = checkout_lock_file(self.lock_file_name()) - self.has_lock = True - return lock_fid, key - return None, None + me = threading.get_ident() + if self.has_lock and self._lock_owner == me: + return None, None + self._thread_lock.acquire() + try: + # Expire after 20 s, as MATLAB's binaryTable/getLock does. A + # table operation takes milliseconds; the expiry only matters + # when a holder dies with the lock (a viewer quit while a + # daemon thread was adding to the cache), and the default hour + # locked every other process out of the cache for that long. + # For the same reason a waiter looks again often (at most every + # 0.1 s), not once a second. + lock_fid, key = checkout_lock_file( + self.lock_file_name(), + check_loops=30, + expiration=20, + max_poll_interval=0.1, + ) + except BaseException: + self._thread_lock.release() + raise + self.has_lock = True + self._lock_owner = me + return lock_fid, key def release_lock(self, lock_fid, key): if key: - release_lock_file(self.lock_file_name(), key) + try: + release_lock_file(self.lock_file_name(), key) + finally: + self._release_thread_lock() + + def _release_thread_lock(self): + """Drop this thread's hold on the table, if it has one.""" + if self.has_lock and self._lock_owner == threading.get_ident(): self.has_lock = False + self._lock_owner = None + self._thread_lock.release() def reset_lock_state(self): """Force ``has_lock`` back to False without touching the disk lock. @@ -570,10 +624,16 @@ def reset_lock_state(self): it, is left alone; the next :meth:`get_lock` re-enters :func:`checkout_lock_file` and resolves real contention normally. + Only the thread that holds the table can reset it: a hold another + thread has is a real one, not a stuck flag, and is left alone. + Mirrors MATLAB ``binaryTable/resetLockState`` (DID-matlab 4a0f9a5). See DID-python#66. """ - self.has_lock = False + if self._lock_owner is None: + self.has_lock = False + else: + self._release_thread_lock() def lock_file_name(self): # "-lock", not ".lock": this is the name MATLAB's binaryTable checks @@ -780,11 +840,16 @@ def find_row( lock_fid, key = (None, None) if is_recurrent else self.get_lock() try: if not sorted: + # One read of the whole column, not one open per row: the + # file cache's index holds a row per cached file, and a + # lookup that reopened it per row took seconds at tens of + # thousands of files. r, _, _ = self.get_size() - for i in range(1, r + 1): - if self.read_row(i, col) == value: - row = i - break + if r: + for i, here in enumerate(self.read_row(None, col), start=1): + if here == value: + row = i + break return row, would_be r_total, _, _ = self.get_size() @@ -1044,8 +1109,14 @@ def get_properties(self): # -- contents --------------------------------------------------------- def is_file(self, file_name_in_cache): - """Is a file with this name in the cache?""" - row, _ = self._table().find_row(1, file_name_in_cache) + """Is a file with this name in the cache? + + A binary search: the index is kept sorted by name (resize_and_add + inserts in order and rewrites it in order), and a linear scan here, + once per file fetched, grew with the size of the cache. MATLAB's + isFile still scans. + """ + row, _ = self._table().find_row(1, file_name_in_cache, sorted=True) return row > 0 def file_list(self, use_catalog=True): @@ -1112,7 +1183,7 @@ def add_file(self, full_path_file_name, file_name_in_cache=None, copy=False): lock_fid, key = table.get_lock() try: destination = self.full_path(file_name_in_cache) - row, _ = table.find_row(1, file_name_in_cache) + row, _ = table.find_row(1, file_name_in_cache, sorted=True) if row: if os.path.isfile(destination): raise ValueError( @@ -1213,7 +1284,7 @@ def touch(self, file_name): table = self._table() lock_fid, key = table.get_lock() try: - row, _ = table.find_row(1, file_name) + row, _ = table.find_row(1, file_name, sorted=True) if not row: return False table.write_entry(row, 2, datenum()) diff --git a/tests/test_file_cache.py b/tests/test_file_cache.py index ebe60c6..fb2a2dc 100644 --- a/tests/test_file_cache.py +++ b/tests/test_file_cache.py @@ -388,3 +388,171 @@ def test_it_forces_has_lock_back_to_false(self, cache, tmp_path): if key: release_lock_file(table.lock_file_name(), key) + + +class TestManyThreadsOneCache: + """get_cache() is one object per process, so threads share its table. + + A worker pool fetching files into the cache adds from many threads at + once. The in-memory has_lock flag was one per object, so while one + thread held the lock every other thread took the "already held" path + and rewrote the index unlocked: rows went missing. + """ + + def test_concurrent_adds_all_land_in_the_index(self, cache, tmp_path): + import threading + + n = 24 + sources = [make_source(tmp_path, i, 10) for i in range(n)] + start = threading.Barrier(n) + errors = [] + + def add(i): + try: + start.wait() + cache.add_file(sources[i], name_of(i)) + except Exception as exc: # noqa: BLE001 - reported below + errors.append(exc) + + threads = [threading.Thread(target=add, args=(i,)) for i in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [] + names, _sizes, _ = cache.file_list() + assert sorted(names) == [name_of(i) for i in range(n)] + assert names == sorted(names) # the index stays sorted + assert cache.get_properties()["currentSize"] == 10 * n + assert all(cache.is_file(name_of(i)) for i in range(n)) + assert cache._table().has_lock is False + + def test_a_held_lock_is_handed_on_quickly(self, cache): + # A waiter used to sleep a whole second between tries; a lock held + # for a few milliseconds cost every waiter a second. + import threading + import time + + table = cache._table() + held = threading.Event() + + def hold_briefly(): + fid, key = table.get_lock() + held.set() + time.sleep(0.05) + table.release_lock(fid, key) + + t = threading.Thread(target=hold_briefly) + t.start() + held.wait() + t0 = time.monotonic() + fid, key = table.get_lock() + waited = time.monotonic() - t0 + table.release_lock(fid, key) + t.join() + assert waited < 0.5 + + def test_another_processes_lock_is_waited_for_not_slept_on(self, cache): + # Contention between processes goes through the lock file alone. + import threading + import time + + from did.file import checkout_lock_file, release_lock_file + + table = cache._table() + _fid, key = checkout_lock_file(table.lock_file_name()) + threading.Timer(0.05, release_lock_file, (table.lock_file_name(), key)).start() + t0 = time.monotonic() + fid, key2 = table.get_lock() + waited = time.monotonic() - t0 + table.release_lock(fid, key2) + assert waited < 0.5 + + +class TestLookupsAreSearches: + def test_is_file_finds_every_member_and_nothing_else(self, cache, tmp_path): + for i in (5, 1, 9, 3, 7): + cache.add_file(make_source(tmp_path, i, 10), name_of(i)) + for i in range(11): + assert cache.is_file(name_of(i)) is (i in (1, 3, 5, 7, 9)) + + def test_an_unsorted_find_reads_the_index_once(self, cache, tmp_path, monkeypatch): + for i in (2, 4, 6): + cache.add_file(make_source(tmp_path, i, 10), name_of(i)) + table = cache._table() + opens = [] + real_open = open + + def counting_open(path, *a, **k): + if str(path) == table.file.fullpathfilename: + opens.append(path) + return real_open(path, *a, **k) + + monkeypatch.setattr("builtins.open", counting_open) + row, _ = table.find_row(1, name_of(6)) + assert row == 3 + assert len(opens) == 1 + + +class TestALockLeftByADeadHolderExpires: + def test_the_table_lock_expires_in_seconds_like_matlabs(self, cache): + # MATLAB's binaryTable/getLock asks for a 20 s expiry. A holder + # killed mid-update (a daemon thread at interpreter exit) must not + # lock every other process out of the cache for an hour. + import datetime as _dt + + from did.file import parse_lock_expiration + + table = cache._table() + fid, key = table.get_lock() + try: + with open(table.lock_file_name()) as f: + expires = parse_lock_expiration(f.readline()) + now = _dt.datetime.now(_dt.timezone.utc).replace(tzinfo=None) + if expires.tzinfo is not None: + expires = expires.astimezone(_dt.timezone.utc).replace(tzinfo=None) + assert 0 < (expires - now).total_seconds() <= 21 + finally: + table.release_lock(fid, key) + assert not os.path.exists(table.lock_file_name()) + + +class TestHowOftenAWaiterLooksAgain: + """Fast polling is opt-in: the cache asks for it, other callers do not.""" + + def _hold_then_release(self, lock_file, after): + import threading + + from did.file import checkout_lock_file, release_lock_file + + _fid, key = checkout_lock_file(lock_file) + threading.Timer(after, release_lock_file, (lock_file, key)).start() + + def test_a_caller_that_asks_is_handed_a_brief_lock_quickly(self, tmp_path): + import time + + from did.file import checkout_lock_file, release_lock_file + + lock_file = str(tmp_path / "thing-lock") + self._hold_then_release(lock_file, 0.05) + t0 = time.monotonic() + _fid, key = checkout_lock_file(lock_file, check_loops=5, max_poll_interval=0.1) + waited = time.monotonic() - t0 + release_lock_file(lock_file, key) + assert waited < 0.5 + + def test_by_default_a_waiter_still_looks_once_a_second(self, tmp_path): + # Locks held for a long time (a database adding thousands of + # documents) keep the old cadence unless their caller says otherwise. + import time + + from did.file import checkout_lock_file, release_lock_file + + lock_file = str(tmp_path / "thing-lock") + self._hold_then_release(lock_file, 0.05) + t0 = time.monotonic() + _fid, key = checkout_lock_file(lock_file, check_loops=5) + waited = time.monotonic() - t0 + release_lock_file(lock_file, key) + assert 0.9 <= waited < 2.0