trying something

Author: bruAristimunha

mne_bids/_fileio.py

@@ -7,14 +7,35 @@
 
 import contextlib
 import inspect
+import os
+import threading
 from contextlib import contextmanager
 from pathlib import Path
 
 from mne.utils import _soft_import, logger, warn
 
+_LOCK_TIMEOUT_FALLBACK = 60.0
+_env_lock_timeout = os.getenv("MNE_BIDS_FILELOCK_TIMEOUT", "")
+try:
+    DEFAULT_LOCK_TIMEOUT = (
+        float(_env_lock_timeout) if _env_lock_timeout else _LOCK_TIMEOUT_FALLBACK
+    )
+    if DEFAULT_LOCK_TIMEOUT <= 0:
+        raise ValueError
+except ValueError:
+    DEFAULT_LOCK_TIMEOUT = _LOCK_TIMEOUT_FALLBACK
+
+_ACTIVE_LOCKS: dict[str, int] = {}
+_ACTIVE_LOCKS_GUARD = threading.RLock()
+
+
+def _canonical_lock_path(path: str | os.PathLike[str]) -> Path:
+    """Return an absolute, normalised path without requiring it to exist."""
+    return Path(path).expanduser().resolve(strict=False)
+
 
 @contextmanager
-def _get_lock_context(path, timeout=5):
+def _get_lock_context(path, timeout=None):
     """Get a file lock context for the given path.
 
     Internal helper function that creates a FileLock if available,
@@ -25,19 +46,27 @@ def _get_lock_context(path, timeout=5):
     path : str
         The path to acquire a lock for.
     timeout : float
-        Timeout in seconds for acquiring the lock (default 60).
+        Timeout in seconds for acquiring the lock. If ``None``, the value of
+        ``DEFAULT_LOCK_TIMEOUT`` (default 60 seconds) is used. The timeout can
+        be overridden via the ``MNE_BIDS_FILELOCK_TIMEOUT`` environment
+        variable.
 
     Yields
     ------
     context : context manager
         Either a FileLock or nullcontext.
     """
+    if timeout is None:
+        timeout = DEFAULT_LOCK_TIMEOUT
+
+    canonical_path = _canonical_lock_path(path)
+
     filelock = _soft_import(
         "filelock", purpose="parallel file I/O locking", strict=False
     )
 
-    lock_path = f"{path}.lock"
-    lock_context = contextlib.nullcontext(enter_result=path)
+    lock_path = canonical_path.with_name(f"{canonical_path.name}.lock")
+    lock_context = contextlib.nullcontext()
 
     stack = "unknown"
     try:  # this should always work but let's be safe
@@ -51,88 +80,118 @@ def _get_lock_context(path, timeout=5):
         del where
     except Exception:
         pass
-    logger.debug(f"Lock: acquiring {path} from {stack}")
+    logger.debug(f"Lock: acquiring {canonical_path} from {stack}")
 
     if filelock:
         try:
             # Ensure parent directory exists
             Path(lock_path).parent.mkdir(parents=True, exist_ok=True)
-            lock_context = filelock.FileLock(lock_path, timeout=timeout)
+            lock_context = filelock.FileLock(
+                str(lock_path),
+                timeout=timeout,
+            )
         except (OSError, TypeError):
             # OSError: permission issues creating lock file
             # TypeError: invalid timeout parameter
             warn("Could not create lock. Proceeding without a lock.")
     try:
         yield lock_context
     except Exception:
-        logger.debug(f"Lock: exception {path} from {stack}")
+        logger.debug(f"Lock: exception {canonical_path} from {stack}")
         raise
     finally:
-        logger.debug(f"Lock: released  {path} from {stack}")
+        logger.debug(f"Lock: released  {canonical_path} from {stack}")
 
 
 @contextmanager
-def _open_lock(path, *args, **kwargs):
+def _open_lock(path, *args, lock_timeout=None, **kwargs):
     """Context manager that acquires a file lock with optional file opening.
 
     If the `filelock` package is available, a lock is acquired on a lock file
     based on the given path (by appending '.lock'). Lock files are left behind
     to avoid race conditions during concurrent operations.
 
+    The lock is re-entrant per process: nested calls for the same ``path`` will
+    reuse the existing lock instead of attempting to acquire it again.
+
     If file opening arguments (*args, **kwargs) are provided, the file is opened
     in the specified mode. Otherwise, just the lock is acquired.
 
     Parameters
     ----------
     path : str
         The path to acquire a lock for (and optionally open).
-    *args, **kwargs : optional
-        Additional arguments and keyword arguments to be passed to the
-        `open` function. If provided, the file will be opened. If empty,
-        only the lock will be acquired.
+    *args : tuple
+        Additional positional arguments forwarded to ``open``.
+    lock_timeout : float | None
+        Timeout in seconds for acquiring the lock. If ``None``, the default
+        timeout applies.
+    **kwargs : dict
+        Additional keyword arguments forwarded to ``open``.
 
     Yields
     ------
     fid : file object or None
         File object if file opening args were provided, None otherwise.
-
     """
-    with _get_lock_context(path, timeout=5) as lock_context:
-        try:
-            with lock_context:
-                if args or kwargs:
-                    # File opening arguments provided - open the file
-                    with open(path, *args, **kwargs) as fid:
-                        yield fid
-                else:
-                    # No file opening arguments - just yield None
-                    yield None
-        finally:
-            # Lock files are left behind to avoid race conditions with concurrent
-            # processes. They should be cleaned up explicitly after all parallel
-            # operations complete via cleanup_lock_files().
-            cleanup_lock_files(path)
+    canonical_path = _canonical_lock_path(path)
+    lock_key = str(canonical_path)
 
+    with _ACTIVE_LOCKS_GUARD:
+        lock_depth = _ACTIVE_LOCKS.get(lock_key, 0)
+        _ACTIVE_LOCKS[lock_key] = lock_depth + 1
+    is_reentrant = lock_depth > 0
+
+    try:
+        if is_reentrant:
+            if args or kwargs:
+                with open(canonical_path, *args, **kwargs) as fid:
+                    yield fid
+            else:
+                yield None
+            return
+
+        with _get_lock_context(
+            canonical_path,
+            timeout=lock_timeout,
+        ) as lock_context:
+            try:
+                with lock_context:
+                    if args or kwargs:
+                        with open(canonical_path, *args, **kwargs) as fid:
+                            yield fid
+                    else:
+                        yield None
+            finally:
+                cleanup_lock_files(canonical_path)
+    finally:
+        with _ACTIVE_LOCKS_GUARD:
+            _ACTIVE_LOCKS[lock_key] -= 1
+            if _ACTIVE_LOCKS[lock_key] == 0:
+                del _ACTIVE_LOCKS[lock_key]
 
-def cleanup_lock_files(root_path):
-    """Remove all .lock files in a directory tree.
 
-    This function should be called after parallel operations complete to clean up
-    lock files that may have been left behind.
+def cleanup_lock_files(root_path):
+    """Remove lock files associated with a path or an entire tree.
 
     Parameters
     ----------
     root_path : str | Path
-        Root directory to search for .lock files.
+        Root directory or file path used to derive lock file locations.
     """
     root_path = Path(root_path)
-    if not root_path.exists():
+
+    if root_path.is_dir():
+        for lock_file in root_path.rglob("*.lock"):
+            try:
+                lock_file.unlink()
+            except OSError:
+                pass
         return
 
-    # Find and remove all .lock files
-    for lock_file in root_path.rglob("*.lock"):
+    lock_candidate = root_path.parent / f"{root_path.name}.lock"
+    if lock_candidate.exists():
         try:
-            lock_file.unlink()
+            lock_candidate.unlink()
         except OSError:
-            # If we can't remove it, skip it (might be in use by another process)
             pass

mne_bids/tests/test_write.py

@@ -106,7 +106,7 @@ def _make_parallel_raw(subject, *, seed=None):
     rng = np.random.default_rng(rng_seed)
     info = mne.create_info(["MEG0113"], 100, ch_types="mag")
     data = rng.standard_normal((1, 100)) * 1e-12
-    raw = mne.io.RawArray(data, info)
+    raw = mne.io.RawArray(data, info, verbose=False)
     raw.set_meas_date(datetime(2020, 1, 1, tzinfo=timezone.utc))
     raw.info["line_freq"] = 60
     raw.info["subject_info"] = {

mne_bids/write.py

@@ -637,7 +637,7 @@ def _participants_tsv(raw, subject_id, fname, overwrite=False):
             if existing_participants:
                 data = _combine_rows(orig_data, data, "participant_id")
 
-        _write_tsv(fname, data)
+        _write_tsv(fname, data, overwrite=True)
 
 
 def _participants_json(fname, overwrite=False):
@@ -819,7 +819,7 @@ def _scans_tsv(raw, raw_fname, fname, keep_source, overwrite=False):
             # otherwise add the new data
             data = _combine_rows(orig_data, data, "filename")
 
-        _write_tsv(fname, data)
+        _write_tsv(fname, data, overwrite=True)
 
 
 def _load_image(image, name="image"):