diff --git a/.github/workflows/prechecks.yml b/.github/workflows/prechecks.yml index 6540983f0e2f1e..a1609044d890dc 100644 --- a/.github/workflows/prechecks.yml +++ b/.github/workflows/prechecks.yml @@ -40,7 +40,7 @@ jobs: # Check that __slots__ are used properly - name: slotscheck run: | - ./bin/spack python -m slotscheck --exclude-modules="spack.test|spack.vendor|spack.new_installer_windows" lib/spack/spack/ + ./bin/spack python -m slotscheck --exclude-modules="spack.test|spack.vendor|spack.new_installer_windows|spack.util.lock_windows" lib/spack/spack/ # Run style checks on the files that have been changed style: diff --git a/etc/spack/defaults/windows/config.yaml b/etc/spack/defaults/windows/config.yaml index 0678fe48b26a3f..230cb55aae2edf 100644 --- a/etc/spack/defaults/windows/config.yaml +++ b/etc/spack/defaults/windows/config.yaml @@ -1,5 +1,4 @@ config: - locks: false build_stage:: - '$user_cache_path/stage' stage_name: '{name}-{version}-{hash:7}' diff --git a/lib/spack/docs/conf.py b/lib/spack/docs/conf.py index fd272146d6ba89..76e6fb7cc1ba24 100644 --- a/lib/spack/docs/conf.py +++ b/lib/spack/docs/conf.py @@ -97,6 +97,7 @@ "_spack_root/lib/spack/spack/test", "_spack_root/lib/spack/spack/package.py", "_spack_root/lib/spack/spack/new_installer_windows.py", + "_spack_root/lib/spack/spack/util/lock_windows.py", ] ) sphinx_apidoc( diff --git a/lib/spack/spack/test/conftest.py b/lib/spack/spack/test/conftest.py index aaf22c7f242c92..cf2cf628e40757 100644 --- a/lib/spack/spack/test/conftest.py +++ b/lib/spack/spack/test/conftest.py @@ -539,8 +539,11 @@ def ignore_stage_files(): Used to track which leftover files in the stage have been seen. """ - # to start with, ignore the .lock file at the stage root. - return set([".lock", spack.stage._source_path_subdir, "build_cache"]) + # to start with, ignore the .lock file at the stage root, and its Windows-only gate lock + # sidecar (see spack.util.lock_windows.WindowsBackend._upgrade_to_write). Neither is ever + # explicitly unlinked for stage locks; they're only ever removed along with the whole stage + # directory. + return set([".lock", ".lock.gate_lock", spack.stage._source_path_subdir, "build_cache"]) def remove_whatever_it_is(path): diff --git a/lib/spack/spack/test/util/lock_backend.py b/lib/spack/spack/test/util/lock_backend.py new file mode 100644 index 00000000000000..9f26fa8e917053 --- /dev/null +++ b/lib/spack/spack/test/util/lock_backend.py @@ -0,0 +1,1444 @@ +# Copyright Spack Project Developers. See COPYRIGHT file for details. +# +# SPDX-License-Identifier: (Apache-2.0 OR MIT) + +"""These tests exercise ``spack.util.lock.Lock`` against whichever real backend +(``PosixBackend``/``WindowsBackend``) the current platform dispatches to. Every test here is +written to run identically -- and hold the same guarantees -- on POSIX and Windows; anything +that only makes sense on one platform belongs in ``lock_unix.py``/``lock_windows.py`` instead. + +Run with pytest:: + + pytest lib/spack/spack/test/util/lock_backend.py + +You can use this to test whether your shared filesystem properly supports reader-writer locking +with byte ranges. + +If you want to test on multiple filesystems, you can modify the ``locations`` list below. By +default it looks like this:: + + locations = [ + tempfile.gettempdir(), # standard tmp directory (potentially local) + '/nfs/tmp2/%u', # NFS tmp mount + '/p/lscratch*/%u' # Lustre scratch mount + ] + +Add names and paths for your preferred filesystem mounts to test on them; the tests are +parametrized to run on all the filesystems listed in this dict. + +""" + +import collections +import getpass +import glob +import multiprocessing +import os +import pathlib +import shutil +import socket +import stat +import sys +import tempfile +import traceback +from contextlib import contextmanager +from multiprocessing import Barrier, Process, Queue + +import pytest + +import spack.util.lock as lk +from spack.util.filesystem import getuid, touch, working_dir + +# +# This test can be run with MPI. MPI is "enabled" if we can import +# mpi4py and the number of total MPI processes is greater than 1. +# Otherwise it just runs as a node-local test. +# +# NOTE: MPI mode is different from node-local mode in that node-local +# mode will spawn its own test processes, while MPI mode assumes you've +# run this script as a SPMD application. In MPI mode, no additional +# processes are spawned, and you need to ensure that you mpirun the +# script with enough processes for all the multiproc_test cases below. +# +# If you don't run with enough processes, tests that require more +# processes than you currently have will be skipped. +# +mpi = False +comm = None +try: + from mpi4py import MPI + + comm = MPI.COMM_WORLD + if comm.size > 1: + mpi = True +except ImportError: + pass + + +#: This is a list of filesystem locations to test locks in. Paths are +#: expanded so that %u is replaced with the current username. '~' is also +#: legal and will be expanded to the user's home directory. +#: +#: Tests are skipped for directories that don't exist, so you'll need to +#: update this with the locations of NFS, Lustre, and other mounts on your +#: system. +locations = [ + tempfile.gettempdir(), + os.path.join("/nfs/tmp2/", getpass.getuser()), + os.path.join("/p/lscratch*/", getpass.getuser()), +] + +#: This is the longest a failed multiproc test will take. +#: Barriers will time out and raise an exception after this interval. +#: In MPI mode, barriers don't time out (they hang). See mpi_multiproc_test. +barrier_timeout = 5 + +#: This is the lock timeout for expected failures. +#: This may need to be higher for some filesystems. +lock_fail_timeout = 0.1 + + +def make_readable(*paths): + # TODO: From os.chmod doc: + # "Note Although Windows supports chmod(), you can only + # set the file's read-only flag with it (via the stat.S_IWRITE and + # stat.S_IREAD constants or a corresponding integer value). All other + # bits are ignored." + for path in paths: + if sys.platform != "win32": + mode = 0o555 if os.path.isdir(path) else 0o444 + else: + mode = stat.S_IREAD + os.chmod(path, mode) + + +def make_writable(*paths): + for path in paths: + if sys.platform != "win32": + mode = 0o755 if os.path.isdir(path) else 0o744 + else: + mode = stat.S_IWRITE + os.chmod(path, mode) + + +@contextmanager +def read_only(*paths): + modes = [os.stat(p).st_mode for p in paths] + make_readable(*paths) + + yield + + for path, mode in zip(paths, modes): + os.chmod(path, mode) + + +@pytest.fixture(scope="session", params=locations) +def lock_test_directory(request): + """This fixture causes tests to be executed for many different mounts. + + See the ``locations`` dict above for details. + """ + return request.param + + +@pytest.fixture(scope="session") +def lock_dir(lock_test_directory): + parent = next( + (p for p in glob.glob(lock_test_directory) if os.path.exists(p) and os.access(p, os.W_OK)), + None, + ) + if not parent: + # Skip filesystems that don't exist or aren't writable + pytest.skip("requires filesystem: '%s'" % lock_test_directory) + elif mpi and parent == tempfile.gettempdir(): + # Skip local tmp test for MPI runs + pytest.skip("skipping local tmp directory for MPI test.") + + tempdir = None + if not mpi or comm.rank == 0: + tempdir = tempfile.mkdtemp(dir=parent) + if mpi: + tempdir = comm.bcast(tempdir) + + yield tempdir + + if mpi: + # rank 0 may get here before others, in which case it'll try to + # remove the directory while other processes try to re-create the + # lock. This will give errno 39: directory not empty. Use a + # barrier to ensure everyone is done first. + comm.barrier() + + if not mpi or comm.rank == 0: + make_writable(tempdir) + shutil.rmtree(tempdir) + + +@pytest.fixture +def private_lock_path(lock_dir): + """In MPI mode, this is a private lock for each rank in a multiproc test. + + For other modes, it is the same as a shared lock. + """ + lock_file = os.path.join(lock_dir, "lockfile") + if mpi: + lock_file += ".%s" % comm.rank + + yield lock_file + + if os.path.exists(lock_file): + make_writable(lock_dir, lock_file) + os.unlink(lock_file) + + +@pytest.fixture +def lock_path(lock_dir): + """This lock is shared among all processes in a multiproc test.""" + lock_file = os.path.join(lock_dir, "lockfile") + + yield lock_file + + if os.path.exists(lock_file): + make_writable(lock_dir, lock_file) + os.unlink(lock_file) + + +def test_poll_interval_generator(): + interval_iter = iter(lk.Lock._poll_interval_generator(_wait_times=[1, 2, 3])) + intervals = [next(interval_iter) for i in range(100)] + assert intervals == [1] * 20 + [2] * 40 + [3] * 40 + + +def local_multiproc_test(*functions, **kwargs): + """Order some processes using simple barrier synchronization.""" + b = Barrier(len(functions), timeout=barrier_timeout) + + args = (b,) + tuple(kwargs.get("extra_args", ())) + procs = [Process(target=f, args=args, name=f.__name__) for f in functions] + + for p in procs: + p.start() + + for p in procs: + p.join() + + assert all(p.exitcode == 0 for p in procs) + + +def mpi_multiproc_test(*functions): + """SPMD version of multiproc test. + + This needs to be run like so: + + srun spack test lock + + Each process executes its corresponding function. This is different + from ``multiproc_test`` above, which spawns the processes. This will + skip tests if there are too few processes to run them. + """ + procs = len(functions) + if procs > comm.size: + pytest.skip("requires at least %d MPI processes" % procs) + + comm.Barrier() # barrier before each MPI test + + include = comm.rank < len(functions) + subcomm = comm.Split(include) + + class subcomm_barrier: + """Stand-in for multiproc barrier for MPI-parallel jobs.""" + + def wait(self): + subcomm.Barrier() + + if include: + try: + functions[subcomm.rank](subcomm_barrier()) + except BaseException: + # aborting is the best we can do for MPI tests without + # hanging, since we're using MPI barriers. This will fail + # early and it loses the nice pytest output, but at least it + # gets use a stacktrace on the processes that failed. + traceback.print_exc() + comm.Abort() + subcomm.Free() + + comm.Barrier() # barrier after each MPI test. + + +#: ``multiproc_test()`` should be called by tests below. +#: ``multiproc_test()`` will work for either MPI runs or for local runs. +multiproc_test = mpi_multiproc_test if mpi else local_multiproc_test + + +# +# Process snippets below can be composed into tests. +# +class AcquireWrite: + def __init__(self, lock_path, start=0, length=1): + self.lock_path = lock_path + self.start = start + self.length = length + + @property + def __name__(self): + return self.__class__.__name__ + + def __call__(self, barrier): + lock = lk.Lock(self.lock_path, start=self.start, length=self.length) + lock.acquire_write() # grab exclusive lock + barrier.wait() + barrier.wait() # hold the lock until timeout in other procs. + + +class AcquireRead: + def __init__(self, lock_path, start=0, length=1): + self.lock_path = lock_path + self.start = start + self.length = length + + @property + def __name__(self): + return self.__class__.__name__ + + def __call__(self, barrier): + lock = lk.Lock(self.lock_path, start=self.start, length=self.length) + lock.acquire_read() # grab shared lock + barrier.wait() + barrier.wait() # hold the lock until timeout in other procs. + + +class TimeoutWrite: + def __init__(self, lock_path, start=0, length=1): + self.lock_path = lock_path + self.start = start + self.length = length + + @property + def __name__(self): + return self.__class__.__name__ + + def __call__(self, barrier): + lock = lk.Lock(self.lock_path, start=self.start, length=self.length) + barrier.wait() # wait for lock acquire in first process + with pytest.raises(lk.LockTimeoutError): + lock.acquire_write(lock_fail_timeout) + barrier.wait() + + +class TimeoutRead: + def __init__(self, lock_path, start=0, length=1): + self.lock_path = lock_path + self.start = start + self.length = length + + @property + def __name__(self): + return self.__class__.__name__ + + def __call__(self, barrier): + lock = lk.Lock(self.lock_path, start=self.start, length=self.length) + barrier.wait() # wait for lock acquire in first process + with pytest.raises(lk.LockTimeoutError): + lock.acquire_read(lock_fail_timeout) + barrier.wait() + + +# +# Test that exclusive locks on other processes time out when an +# exclusive lock is held. +# +def test_write_lock_timeout_on_write(lock_path): + multiproc_test(AcquireWrite(lock_path), TimeoutWrite(lock_path)) + + +def test_write_lock_timeout_on_write_2(lock_path): + multiproc_test(AcquireWrite(lock_path), TimeoutWrite(lock_path), TimeoutWrite(lock_path)) + + +def test_write_lock_timeout_on_write_3(lock_path): + multiproc_test( + AcquireWrite(lock_path), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + ) + + +def test_write_lock_timeout_on_write_ranges(lock_path): + multiproc_test(AcquireWrite(lock_path, 0, 1), TimeoutWrite(lock_path, 0, 1)) + + +def test_write_lock_timeout_on_write_ranges_2(lock_path): + multiproc_test( + AcquireWrite(lock_path, 0, 64), + AcquireWrite(lock_path, 65, 1), + TimeoutWrite(lock_path, 0, 1), + TimeoutWrite(lock_path, 63, 1), + ) + + +def test_write_lock_timeout_on_write_ranges_3(lock_path): + multiproc_test( + AcquireWrite(lock_path, 0, 1), + AcquireWrite(lock_path, 1, 1), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + ) + + +def test_write_lock_timeout_on_write_ranges_4(lock_path): + multiproc_test( + AcquireWrite(lock_path, 0, 1), + AcquireWrite(lock_path, 1, 1), + AcquireWrite(lock_path, 2, 456), + AcquireWrite(lock_path, 500, 64), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + ) + + +# +# Test that shared locks on other processes time out when an +# exclusive lock is held. +# +def test_read_lock_timeout_on_write(lock_path): + multiproc_test(AcquireWrite(lock_path), TimeoutRead(lock_path)) + + +def test_read_lock_timeout_on_write_2(lock_path): + multiproc_test(AcquireWrite(lock_path), TimeoutRead(lock_path), TimeoutRead(lock_path)) + + +def test_read_lock_timeout_on_write_3(lock_path): + multiproc_test( + AcquireWrite(lock_path), + TimeoutRead(lock_path), + TimeoutRead(lock_path), + TimeoutRead(lock_path), + ) + + +def test_read_lock_timeout_on_write_ranges(lock_path): + """small write lock, read whole file.""" + multiproc_test(AcquireWrite(lock_path, 0, 1), TimeoutRead(lock_path)) + + +def test_read_lock_timeout_on_write_ranges_2(lock_path): + """small write lock, small read lock""" + multiproc_test(AcquireWrite(lock_path, 0, 1), TimeoutRead(lock_path, 0, 1)) + + +def test_read_lock_timeout_on_write_ranges_3(lock_path): + """two write locks, overlapping read locks""" + multiproc_test( + AcquireWrite(lock_path, 0, 1), + AcquireWrite(lock_path, 64, 128), + TimeoutRead(lock_path, 0, 1), + TimeoutRead(lock_path, 128, 256), + ) + + +# +# Test that exclusive locks time out when shared locks are held. +# +def test_write_lock_timeout_on_read(lock_path): + multiproc_test(AcquireRead(lock_path), TimeoutWrite(lock_path)) + + +def test_write_lock_timeout_on_read_2(lock_path): + multiproc_test(AcquireRead(lock_path), TimeoutWrite(lock_path), TimeoutWrite(lock_path)) + + +def test_write_lock_timeout_on_read_3(lock_path): + multiproc_test( + AcquireRead(lock_path), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + ) + + +def test_write_lock_timeout_on_read_ranges(lock_path): + multiproc_test(AcquireRead(lock_path, 0, 1), TimeoutWrite(lock_path)) + + +def test_write_lock_timeout_on_read_ranges_2(lock_path): + multiproc_test(AcquireRead(lock_path, 0, 1), TimeoutWrite(lock_path, 0, 1)) + + +def test_write_lock_timeout_on_read_ranges_3(lock_path): + multiproc_test( + AcquireRead(lock_path, 0, 1), + AcquireRead(lock_path, 10, 1), + TimeoutWrite(lock_path, 0, 1), + TimeoutWrite(lock_path, 10, 1), + ) + + +def test_write_lock_timeout_on_read_ranges_4(lock_path): + multiproc_test( + AcquireRead(lock_path, 0, 64), + TimeoutWrite(lock_path, 10, 1), + TimeoutWrite(lock_path, 32, 1), + ) + + +def test_write_lock_timeout_on_read_ranges_5(lock_path): + multiproc_test( + AcquireRead(lock_path, 64, 128), + TimeoutWrite(lock_path, 65, 1), + TimeoutWrite(lock_path, 127, 1), + TimeoutWrite(lock_path, 90, 10), + ) + + +# +# Test that exclusive locks time while lots of shared locks are held. +# +def test_write_lock_timeout_with_multiple_readers_2_1(lock_path): + multiproc_test(AcquireRead(lock_path), AcquireRead(lock_path), TimeoutWrite(lock_path)) + + +def test_write_lock_timeout_with_multiple_readers_2_2(lock_path): + multiproc_test( + AcquireRead(lock_path), + AcquireRead(lock_path), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + ) + + +def test_write_lock_timeout_with_multiple_readers_3_1(lock_path): + multiproc_test( + AcquireRead(lock_path), + AcquireRead(lock_path), + AcquireRead(lock_path), + TimeoutWrite(lock_path), + ) + + +def test_write_lock_timeout_with_multiple_readers_3_2(lock_path): + multiproc_test( + AcquireRead(lock_path), + AcquireRead(lock_path), + AcquireRead(lock_path), + TimeoutWrite(lock_path), + TimeoutWrite(lock_path), + ) + + +def test_write_lock_timeout_with_multiple_readers_2_1_ranges(lock_path): + multiproc_test( + AcquireRead(lock_path, 0, 10), AcquireRead(lock_path, 2, 10), TimeoutWrite(lock_path, 5, 5) + ) + + +def test_write_lock_timeout_with_multiple_readers_2_3_ranges(lock_path): + multiproc_test( + AcquireRead(lock_path, 0, 10), + AcquireRead(lock_path, 5, 15), + TimeoutWrite(lock_path, 0, 1), + TimeoutWrite(lock_path, 11, 3), + TimeoutWrite(lock_path, 7, 1), + ) + + +def test_write_lock_timeout_with_multiple_readers_3_1_ranges(lock_path): + multiproc_test( + AcquireRead(lock_path, 0, 5), + AcquireRead(lock_path, 5, 5), + AcquireRead(lock_path, 10, 5), + TimeoutWrite(lock_path, 0, 15), + ) + + +def test_write_lock_timeout_with_multiple_readers_3_2_ranges(lock_path): + multiproc_test( + AcquireRead(lock_path, 0, 5), + AcquireRead(lock_path, 5, 5), + AcquireRead(lock_path, 10, 5), + TimeoutWrite(lock_path, 3, 10), + TimeoutWrite(lock_path, 5, 1), + ) + + +def test_read_lock_read_only_dir_writable_lockfile(lock_dir, lock_path): + """read-only directory, writable lockfile.""" + touch(lock_path) + with read_only(lock_dir): + lock = lk.Lock(lock_path) + + with lk.ReadTransaction(lock): + pass + + with lk.WriteTransaction(lock): + pass + + +def test_upgrade_read_to_write(private_lock_path): + """Test that a read lock can be upgraded to a write lock. + + Note that to upgrade a read lock to a write lock, you have the be the + only holder of a read lock. Client code needs to coordinate that for + shared locks. For this test, we use a private lock just to test that an + upgrade is possible. + """ + # ensure lock file exists the first time, so we open it read-only + # to begin with. + touch(private_lock_path) + + lock = lk.Lock(private_lock_path) + assert lock._reads == 0 + assert lock._writes == 0 + + lock.acquire_read() + assert lock._reads == 1 + assert lock._writes == 0 + assert lock.backend._file_ref.fh.mode == "rb+" + + lock.acquire_write() + assert lock._reads == 1 + assert lock._writes == 1 + assert lock.backend._file_ref.fh.mode == "rb+" + + lock.release_write() + assert lock._reads == 1 + assert lock._writes == 0 + assert lock.backend._file_ref.fh.mode == "rb+" + + lock.release_read() + assert lock._reads == 0 + assert lock._writes == 0 + # On Windows, _unlock() closes the file handle so the file can be deleted + # (Windows raises WinError 32 on unlink if any process has the file open). + if sys.platform != "win32": + assert not lock.backend._file_ref.fh.closed # recycle the file handle for next lock + + +def test_release_write_downgrades_to_shared(private_lock_path): + """Releasing a write lock while a read lock is held must downgrade the POSIX lock + from exclusive to shared, allowing other processes to acquire read locks.""" + lock = lk.Lock(private_lock_path) + lock.acquire_read() + lock.acquire_write() + lock.release_write() + assert lock._reads == 1 + assert lock._writes == 0 + + ctx = multiprocessing.get_context() + q = ctx.Queue() + + # Another process must be able to acquire a shared read lock concurrently. + p = ctx.Process(target=_child_try_acquire_read, args=(private_lock_path, q)) + p.start() + p.join() + assert q.get() is True + + # But must not be able to acquire an exclusive write lock. + p = ctx.Process(target=_child_try_acquire_write, args=(private_lock_path, q)) + p.start() + p.join() + assert q.get() is False + + lock.release_read() + assert lock._reads == 0 + assert lock._writes == 0 + + +@pytest.mark.skipif(getuid() == 0, reason="user is root") +def test_upgrade_read_to_write_fails_with_readonly_file(private_lock_path): + """Test that read-only file can be read-locked but not write-locked.""" + # ensure lock file exists the first time + touch(private_lock_path) + + # open it read-only to begin with. + with read_only(private_lock_path): + lock = lk.Lock(private_lock_path) + assert lock._reads == 0 + assert lock._writes == 0 + + lock.acquire_read() + assert lock._reads == 1 + assert lock._writes == 0 + assert lock.backend._file_ref.fh.mode == "rb" + + # upgrade to write here + with pytest.raises(lk.LockROFileError): + lock.acquire_write() + + lock.release_read() + + +class ComplexAcquireAndRelease: + def __init__(self, lock_path): + self.lock_path = lock_path + + def p1(self, barrier): + lock = lk.Lock(self.lock_path) + + lock.acquire_write() + barrier.wait() # ---------------------------------------- 1 + # others test timeout + barrier.wait() # ---------------------------------------- 2 + lock.release_write() # release and others acquire read + barrier.wait() # ---------------------------------------- 3 + with pytest.raises(lk.LockTimeoutError): + lock.acquire_write(lock_fail_timeout) + lock.acquire_read() + barrier.wait() # ---------------------------------------- 4 + lock.release_read() + barrier.wait() # ---------------------------------------- 5 + + # p2 upgrades read to write + barrier.wait() # ---------------------------------------- 6 + with pytest.raises(lk.LockTimeoutError): + lock.acquire_write(lock_fail_timeout) + with pytest.raises(lk.LockTimeoutError): + lock.acquire_read(lock_fail_timeout) + barrier.wait() # ---------------------------------------- 7 + # p2 releases write and read + barrier.wait() # ---------------------------------------- 8 + + # p3 acquires read + barrier.wait() # ---------------------------------------- 9 + # p3 upgrades read to write + barrier.wait() # ---------------------------------------- 10 + with pytest.raises(lk.LockTimeoutError): + lock.acquire_write(lock_fail_timeout) + with pytest.raises(lk.LockTimeoutError): + lock.acquire_read(lock_fail_timeout) + barrier.wait() # ---------------------------------------- 11 + # p3 releases locks + barrier.wait() # ---------------------------------------- 12 + lock.acquire_read() + barrier.wait() # ---------------------------------------- 13 + lock.release_read() + + def p2(self, barrier): + lock = lk.Lock(self.lock_path) + + # p1 acquires write + barrier.wait() # ---------------------------------------- 1 + with pytest.raises(lk.LockTimeoutError): + lock.acquire_write(lock_fail_timeout) + with pytest.raises(lk.LockTimeoutError): + lock.acquire_read(lock_fail_timeout) + barrier.wait() # ---------------------------------------- 2 + lock.acquire_read() + barrier.wait() # ---------------------------------------- 3 + # p1 tests shared read + barrier.wait() # ---------------------------------------- 4 + # others release reads + barrier.wait() # ---------------------------------------- 5 + + lock.acquire_write() # upgrade read to write + barrier.wait() # ---------------------------------------- 6 + # others test timeout + barrier.wait() # ---------------------------------------- 7 + lock.release_write() # release read AND write (need both) + lock.release_read() + barrier.wait() # ---------------------------------------- 8 + + # p3 acquires read + barrier.wait() # ---------------------------------------- 9 + # p3 upgrades read to write + barrier.wait() # ---------------------------------------- 10 + with pytest.raises(lk.LockTimeoutError): + lock.acquire_write(lock_fail_timeout) + with pytest.raises(lk.LockTimeoutError): + lock.acquire_read(lock_fail_timeout) + barrier.wait() # ---------------------------------------- 11 + # p3 releases locks + barrier.wait() # ---------------------------------------- 12 + lock.acquire_read() + barrier.wait() # ---------------------------------------- 13 + lock.release_read() + + def p3(self, barrier): + lock = lk.Lock(self.lock_path) + + # p1 acquires write + barrier.wait() # ---------------------------------------- 1 + with pytest.raises(lk.LockTimeoutError): + lock.acquire_write(lock_fail_timeout) + with pytest.raises(lk.LockTimeoutError): + lock.acquire_read(lock_fail_timeout) + barrier.wait() # ---------------------------------------- 2 + lock.acquire_read() + barrier.wait() # ---------------------------------------- 3 + # p1 tests shared read + barrier.wait() # ---------------------------------------- 4 + lock.release_read() + barrier.wait() # ---------------------------------------- 5 + + # p2 upgrades read to write + barrier.wait() # ---------------------------------------- 6 + with pytest.raises(lk.LockTimeoutError): + lock.acquire_write(lock_fail_timeout) + with pytest.raises(lk.LockTimeoutError): + lock.acquire_read(lock_fail_timeout) + barrier.wait() # ---------------------------------------- 7 + # p2 releases write & read + barrier.wait() # ---------------------------------------- 8 + + lock.acquire_read() + barrier.wait() # ---------------------------------------- 9 + lock.acquire_write() + barrier.wait() # ---------------------------------------- 10 + # others test timeout + barrier.wait() # ---------------------------------------- 11 + lock.release_read() # release read AND write in opposite + lock.release_write() # order from before on p2 + barrier.wait() # ---------------------------------------- 12 + lock.acquire_read() + barrier.wait() # ---------------------------------------- 13 + lock.release_read() + + +# +# Longer test case that ensures locks are reusable. Ordering is +# enforced by barriers throughout -- steps are shown with numbers. +# +def test_complex_acquire_and_release_chain(lock_path): + test_chain = ComplexAcquireAndRelease(lock_path) + multiproc_test(test_chain.p1, test_chain.p2, test_chain.p3) + + +class AssertLock(lk.Lock): + """Test lock class that marks acquire/release events.""" + + def __init__(self, lock_path, vals): + super().__init__(lock_path) + self.vals = vals + + # assert hooks for subclasses + assert_acquire_read = lambda self: None + assert_acquire_write = lambda self: None + assert_release_read = lambda self: None + assert_release_write = lambda self: None + + def acquire_read(self, timeout=None): + self.assert_acquire_read() + result = super().acquire_read(timeout) + self.vals["acquired_read"] = True + return result + + def acquire_write(self, timeout=None): + self.assert_acquire_write() + result = super().acquire_write(timeout) + self.vals["acquired_write"] = True + return result + + def release_read(self, release_fn=None): + self.assert_release_read() + result = super().release_read(release_fn) + self.vals["released_read"] = True + return result + + def release_write(self, release_fn=None): + self.assert_release_write() + result = super().release_write(release_fn) + self.vals["released_write"] = True + return result + + +@pytest.mark.parametrize( + "transaction,type", [(lk.ReadTransaction, "read"), (lk.WriteTransaction, "write")] +) +def test_transaction(lock_path, transaction, type): + class MockLock(AssertLock): + def assert_acquire_read(self): + assert not vals["entered_fn"] + assert not vals["exited_fn"] + + def assert_release_read(self): + assert vals["entered_fn"] + assert not vals["exited_fn"] + + def assert_acquire_write(self): + assert not vals["entered_fn"] + assert not vals["exited_fn"] + + def assert_release_write(self): + assert vals["entered_fn"] + assert not vals["exited_fn"] + + def enter_fn(): + # assert enter_fn is called while lock is held + assert vals["acquired_%s" % type] + vals["entered_fn"] = True + + def exit_fn(t, v, tb): + # assert exit_fn is called while lock is held + assert not vals["released_%s" % type] + vals["exited_fn"] = True + vals["exception"] = t or v or tb + + vals = collections.defaultdict(lambda: False) + lock = MockLock(lock_path, vals) + + with transaction(lock, acquire=enter_fn, release=exit_fn): + assert vals["acquired_%s" % type] + assert not vals["released_%s" % type] + + assert vals["entered_fn"] + assert vals["exited_fn"] + assert vals["acquired_%s" % type] + assert vals["released_%s" % type] + assert not vals["exception"] + + +@pytest.mark.parametrize( + "transaction,type", [(lk.ReadTransaction, "read"), (lk.WriteTransaction, "write")] +) +def test_transaction_with_exception(lock_path, transaction, type): + class MockLock(AssertLock): + def assert_acquire_read(self): + assert not vals["entered_fn"] + assert not vals["exited_fn"] + + def assert_release_read(self): + assert vals["entered_fn"] + assert not vals["exited_fn"] + + def assert_acquire_write(self): + assert not vals["entered_fn"] + assert not vals["exited_fn"] + + def assert_release_write(self): + assert vals["entered_fn"] + assert not vals["exited_fn"] + + def enter_fn(): + assert vals["acquired_%s" % type] + vals["entered_fn"] = True + + def exit_fn(t, v, tb): + assert not vals["released_%s" % type] + vals["exited_fn"] = True + vals["exception"] = t or v or tb + return exit_result + + exit_result = False + vals = collections.defaultdict(lambda: False) + lock = MockLock(lock_path, vals) + + with pytest.raises(Exception): + with transaction(lock, acquire=enter_fn, release=exit_fn): + raise Exception() + + assert vals["entered_fn"] + assert vals["exited_fn"] + assert vals["exception"] + + # test suppression of exceptions from exit_fn + exit_result = True + vals.clear() + + # should not raise now. + with transaction(lock, acquire=enter_fn, release=exit_fn): + raise Exception() + + assert vals["entered_fn"] + assert vals["exited_fn"] + assert vals["exception"] + + +def test_nested_write_transaction(lock_path): + """Ensure that the outermost write transaction writes.""" + + def write(t, v, tb): + vals["wrote"] = True + + vals = collections.defaultdict(lambda: False) + lock = AssertLock(lock_path, vals) + + # write/write + with lk.WriteTransaction(lock, release=write): + assert not vals["wrote"] + with lk.WriteTransaction(lock, release=write): + assert not vals["wrote"] + assert not vals["wrote"] + assert vals["wrote"] + + # read/write + vals.clear() + with lk.ReadTransaction(lock): + assert not vals["wrote"] + with lk.WriteTransaction(lock, release=write): + assert not vals["wrote"] + assert vals["wrote"] + + # write/read/write + vals.clear() + with lk.WriteTransaction(lock, release=write): + assert not vals["wrote"] + with lk.ReadTransaction(lock): + assert not vals["wrote"] + with lk.WriteTransaction(lock, release=write): + assert not vals["wrote"] + assert not vals["wrote"] + assert not vals["wrote"] + assert vals["wrote"] + + # read/write/read/write + vals.clear() + with lk.ReadTransaction(lock): + with lk.WriteTransaction(lock, release=write): + assert not vals["wrote"] + with lk.ReadTransaction(lock): + assert not vals["wrote"] + with lk.WriteTransaction(lock, release=write): + assert not vals["wrote"] + assert not vals["wrote"] + assert not vals["wrote"] + assert vals["wrote"] + + +def test_nested_reads(lock_path): + """Ensure that write transactions won't re-read data.""" + + def read(): + vals["read"] += 1 + + vals = collections.defaultdict(lambda: 0) + lock = AssertLock(lock_path, vals) + + # read/read + vals.clear() + assert vals["read"] == 0 + with lk.ReadTransaction(lock, acquire=read): + assert vals["read"] == 1 + with lk.ReadTransaction(lock, acquire=read): + assert vals["read"] == 1 + + # write/write + vals.clear() + assert vals["read"] == 0 + with lk.WriteTransaction(lock, acquire=read): + assert vals["read"] == 1 + with lk.WriteTransaction(lock, acquire=read): + assert vals["read"] == 1 + + # read/write + vals.clear() + assert vals["read"] == 0 + with lk.ReadTransaction(lock, acquire=read): + assert vals["read"] == 1 + with lk.WriteTransaction(lock, acquire=read): + assert vals["read"] == 1 + + # write/read/write + vals.clear() + assert vals["read"] == 0 + with lk.WriteTransaction(lock, acquire=read): + assert vals["read"] == 1 + with lk.ReadTransaction(lock, acquire=read): + assert vals["read"] == 1 + with lk.WriteTransaction(lock, acquire=read): + assert vals["read"] == 1 + + # read/write/read/write + vals.clear() + assert vals["read"] == 0 + with lk.ReadTransaction(lock, acquire=read): + assert vals["read"] == 1 + with lk.WriteTransaction(lock, acquire=read): + assert vals["read"] == 1 + with lk.ReadTransaction(lock, acquire=read): + assert vals["read"] == 1 + with lk.WriteTransaction(lock, acquire=read): + assert vals["read"] == 1 + + +@pytest.mark.parametrize( + "transaction,type", [(lk.TryReadTransaction, "read"), (lk.TryWriteTransaction, "write")] +) +def test_try_transaction(lock_path, transaction, type): + """When uncontended, try-transactions yield True and behave like blocking transactions.""" + + def enter_fn(): + vals["entered_fn"] = True + + def exit_fn(t, v, tb): + vals["exited_fn"] = True + vals["exception"] = t or v or tb + + vals = collections.defaultdict(lambda: False) + lock = AssertLock(lock_path, vals) + + with transaction(lock, acquire=enter_fn, release=exit_fn) as acquired: + assert acquired + assert vals["entered_fn"] + assert not vals["released_%s" % type] + + assert vals["exited_fn"] + assert vals["released_%s" % type] + assert not vals["exception"] + + +@pytest.mark.parametrize( + "transaction,attr", + [(lk.TryReadTransaction, "try_acquire_read"), (lk.TryWriteTransaction, "try_acquire_write")], +) +def test_try_transaction_blocked(lock_path, transaction, attr, monkeypatch): + """When the lock would block, try-transactions yield False and call no functions on exit.""" + + def enter_fn(): + vals["entered_fn"] = True + + def exit_fn(t, v, tb): + vals["exited_fn"] = True + + vals = collections.defaultdict(lambda: False) + lock = AssertLock(lock_path, vals) + monkeypatch.setattr(lock, attr, lambda: False) + + with transaction(lock, acquire=enter_fn, release=exit_fn) as acquired: + assert not acquired + + assert not vals["entered_fn"] + assert not vals["exited_fn"] + assert lock._reads == 0 and lock._writes == 0 + + # exceptions raised in the body still propagate + with pytest.raises(ValueError): + with transaction(lock, acquire=enter_fn, release=exit_fn) as acquired: + assert not acquired + raise ValueError() + + +def test_try_transaction_with_exception(lock_path): + """An exception in the body propagates and is forwarded to the release function.""" + + def exit_fn(t, v, tb): + vals["exception"] = t or v or tb + + vals = collections.defaultdict(lambda: False) + lock = AssertLock(lock_path, vals) + + with pytest.raises(ValueError): + with lk.TryWriteTransaction(lock, release=exit_fn) as acquired: + assert acquired + raise ValueError() + + assert vals["exception"] + assert vals["released_write"] + assert lock._reads == 0 and lock._writes == 0 + + +def test_try_transaction_nested(lock_path): + """Nested try-transactions acquire, but do not re-run the acquire function, and the release + function only runs when the outermost write lock is released.""" + + def read(): + vals["read"] += 1 + + def write(t, v, tb): + vals["wrote"] += 1 + + vals = collections.defaultdict(lambda: 0) + lock = AssertLock(lock_path, vals) + + with lk.WriteTransaction(lock, acquire=read, release=write): + assert vals["read"] == 1 + with lk.TryWriteTransaction(lock, acquire=read, release=write) as acquired: + assert acquired + assert vals["read"] == 1 + assert vals["wrote"] == 0 + with lk.TryReadTransaction(lock, acquire=read) as acquired: + assert acquired + assert vals["read"] == 1 + assert lock._writes == 1 + assert vals["wrote"] == 1 + assert lock._reads == 0 and lock._writes == 0 + + +class LockDebugOutput: + def __init__(self, lock_path): + self.lock_path = lock_path + self.host = socket.gethostname() + + def p1(self, barrier, q1, q2): + # exchange pids + p1_pid = os.getpid() + q1.put(p1_pid) + p2_pid = q2.get() + + # set up lock + lock = lk.Lock(self.lock_path, debug=True) + + with lk.WriteTransaction(lock): + # p1 takes write lock and writes pid/host to file + barrier.wait() # ------------------------------------ 1 + + assert lock.backend.pid == p1_pid + assert lock.backend.host == self.host + + # wait for p2 to verify contents of file + barrier.wait() # ---------------------------------------- 2 + + # wait for p2 to take a write lock + barrier.wait() # ---------------------------------------- 3 + + # verify pid/host info again + with lk.ReadTransaction(lock): + assert lock.backend.old_pid == p1_pid + assert lock.backend.old_host == self.host + + assert lock.backend.pid == p2_pid + assert lock.backend.host == self.host + + barrier.wait() # ---------------------------------------- 4 + + def p2(self, barrier, q1, q2): + # exchange pids + p2_pid = os.getpid() + p1_pid = q1.get() + q2.put(p2_pid) + + # set up lock + lock = lk.Lock(self.lock_path, debug=True) + + # p1 takes write lock and writes pid/host to file + barrier.wait() # ---------------------------------------- 1 + + # verify that p1 wrote information to lock file + with lk.ReadTransaction(lock): + assert lock.backend.pid == p1_pid + assert lock.backend.host == self.host + + barrier.wait() # ---------------------------------------- 2 + + # take a write lock on the file and verify pid/host info + with lk.WriteTransaction(lock): + assert lock.backend.old_pid == p1_pid + assert lock.backend.old_host == self.host + + assert lock.backend.pid == p2_pid + assert lock.backend.host == self.host + + barrier.wait() # ------------------------------------ 3 + + # wait for p1 to verify pid/host info + barrier.wait() # ---------------------------------------- 4 + + +def test_lock_debug_output(lock_path): + test_debug = LockDebugOutput(lock_path) + q1, q2 = Queue(), Queue() + local_multiproc_test(test_debug.p2, test_debug.p1, extra_args=(q1, q2)) + + +def test_lock_with_no_parent_directory(tmp_path: pathlib.Path): + """Make sure locks work even when their parent directory does not exist.""" + with working_dir(str(tmp_path)): + lock = lk.Lock("foo/bar/baz/lockfile") + with lk.WriteTransaction(lock): + pass + + +def test_lock_in_current_directory(tmp_path: pathlib.Path): + """Make sure locks work even when their parent directory does not exist.""" + with working_dir(str(tmp_path)): + # test we can create a lock in the current directory + lock = lk.Lock("lockfile") + for i in range(10): + with lk.ReadTransaction(lock): + pass + with lk.WriteTransaction(lock): + pass + + # and that we can do the same thing after it's already there + lock = lk.Lock("lockfile") + for i in range(10): + with lk.ReadTransaction(lock): + pass + with lk.WriteTransaction(lock): + pass + + +def test_attempts_str(): + assert lk._attempts_str(0, 0) == "" + assert lk._attempts_str(0.12, 1) == "" + assert lk._attempts_str(12.345, 2) == " after 12.345s and 2 attempts" + + +def test_lock_str(): + lock = lk.Lock("lockfile") + lockstr = str(lock) + assert f"lockfile[0:{lk.WHOLE_FILE_RANGE}]" in lockstr + assert "timeout=None" in lockstr + assert "#reads=0, #writes=0" in lockstr + + +def test_downgrade_write_okay(tmp_path: pathlib.Path): + """Test the lock write-to-read downgrade operation.""" + with working_dir(str(tmp_path)): + lock = lk.Lock("lockfile") + lock.acquire_write() + lock.downgrade_write_to_read() + assert lock._reads == 1 + assert lock._writes == 0 + lock.release_read() + + +def test_downgrade_write_fails(tmp_path: pathlib.Path): + """Test failing the lock write-to-read downgrade operation.""" + with working_dir(str(tmp_path)): + lock = lk.Lock("lockfile") + lock.acquire_read() + msg = "Cannot downgrade lock from write to read on file: lockfile" + with pytest.raises(lk.LockDowngradeError, match=msg): + lock.downgrade_write_to_read() + lock.release_read() + + +def test_upgrade_read_okay(tmp_path: pathlib.Path): + """Test the lock read-to-write upgrade operation.""" + with working_dir(str(tmp_path)): + lock = lk.Lock("lockfile") + lock.acquire_read() + lock.upgrade_read_to_write() + assert lock._reads == 0 + assert lock._writes == 1 + lock.release_write() + + +def test_upgrade_read_fails(tmp_path: pathlib.Path): + """Test failing the lock read-to-write upgrade operation.""" + with working_dir(str(tmp_path)): + lock = lk.Lock("lockfile") + lock.acquire_write() + msg = "Cannot upgrade lock from read to write on file: lockfile" + with pytest.raises(lk.LockUpgradeError, match=msg): + lock.upgrade_read_to_write() + lock.release_write() + + +def _child_try_acquire_write(lock_path: str, result_queue): + lock = lk.Lock(lock_path) + result_queue.put(lock.try_acquire_write()) + + +def _child_try_acquire_read(lock_path: str, result_queue): + lock = lk.Lock(lock_path) + result_queue.put(lock.try_acquire_read()) + + +def test_try_acquire_read(tmp_path: pathlib.Path): + """Test non-blocking try_acquire_read.""" + lock = lk.Lock(str(tmp_path / "lockfile")) + + # Succeeds on unlocked lock + assert lock.try_acquire_read() is True + assert lock._reads == 1 + + # Succeeds again (nested) + assert lock.try_acquire_read() is True + assert lock._reads == 2 + + lock.release_read() + lock.release_read() + ctx = multiprocessing.get_context() + + # Fails when another process holds an exclusive write lock + lock.acquire_write() + try: + q = ctx.Queue() + p = ctx.Process(target=_child_try_acquire_read, args=(str(tmp_path / "lockfile"), q)) + p.start() + p.join() + assert q.get() is False + finally: + lock.release_write() + + +def test_try_acquire_write(tmp_path: pathlib.Path): + """Test non-blocking try_acquire_write.""" + lock = lk.Lock(str(tmp_path / "lockfile")) + ctx = multiprocessing.get_context() + + # Succeeds on unlocked lock + assert lock.try_acquire_write() is True + assert lock._writes == 1 + + # Succeeds again (nested) + assert lock.try_acquire_write() is True + assert lock._writes == 2 + + lock.release_write() + lock.release_write() + + # Fails when another process holds a write lock + lock.acquire_write() + try: + q = ctx.Queue() + p = ctx.Process(target=_child_try_acquire_write, args=(str(tmp_path / "lockfile"), q)) + p.start() + p.join() + assert q.get() is False + finally: + lock.release_write() + + # Fails when another process holds a read lock + lock.acquire_read() + try: + q = ctx.Queue() + p = ctx.Process(target=_child_try_acquire_write, args=(str(tmp_path / "lockfile"), q)) + p.start() + p.join() + assert q.get() is False + finally: + lock.release_read() + + +def test_try_acquire_write_from_read(tmp_path: pathlib.Path): + """try_acquire_write must properly upgrade an existing read to a write.""" + lock = lk.Lock(str(tmp_path / "lockfile")) + + # Acquire a read, then non-blockingly upgrade to write + lock.acquire_read() + assert lock._reads == 1 + assert lock._writes == 0 + + assert lock.try_acquire_write() is True + assert lock._reads == 0 + assert lock._writes == 1 + + # Nested try_acquire_write while holding write + assert lock.try_acquire_write() is True + assert lock._writes == 2 + + lock.release_write() + assert lock._writes == 1 + lock.release_write() + assert lock._reads == 0 + assert lock._writes == 0 + + +def _child_fails_to_acquire_read(_lock: lk.Lock): + try: + _lock.acquire_read(timeout=1e-9) + except lk.LockTimeoutError: + return + assert False, "Child process should not have been able to acquire read lock" + + +def test_read_after_write_does_not_accidentally_downgrade(tmp_path: pathlib.Path): + """Test that acquiring a read lock after a write lock does not accidentally downgrade the + write lock, by having another process attempt to acquire a read lock.""" + lock = lk.Lock(str(tmp_path / "lockfile")) + lock.acquire_write() + lock.acquire_read() # should not downgrade the write lock + try: + # No matter the start method, the child process shouldn't be able to acquire a read lock. + p = multiprocessing.Process(target=_child_fails_to_acquire_read, args=(lock,)) + p.start() + p.join() + assert p.exitcode == 0 + finally: + lock.release_read() + lock.release_write() diff --git a/lib/spack/spack/test/util/lock_unix.py b/lib/spack/spack/test/util/lock_unix.py index 1f7bd550fd381a..7fe81b33082d6c 100644 --- a/lib/spack/spack/test/util/lock_unix.py +++ b/lib/spack/spack/test/util/lock_unix.py @@ -2,572 +2,35 @@ # # SPDX-License-Identifier: (Apache-2.0 OR MIT) -"""These tests ensure that our lock works correctly. +"""Tests for behavior specific to ``spack.util.lock_posix.PosixBackend`` (``fcntl``-based locking) +that can't be exercised through the shared, platform-neutral tests in ``lock_backend.py``. Run with pytest:: - pytest lib/spack/spack/test/util/lock.py - -You can use this to test whether your shared filesystem properly supports -POSIX reader-writer locking with byte ranges through fcntl. - -If you want to test on multiple filesystems, you can modify the -``locations`` list below. By default it looks like this:: - - locations = [ - tempfile.gettempdir(), # standard tmp directory (potentially local) - '/nfs/tmp2/%u', # NFS tmp mount - '/p/lscratch*/%u' # Lustre scratch mount - ] - -Add names and paths for your preferred filesystem mounts to test on them; -the tests are parametrized to run on all the filesystems listed in this -dict. - + pytest lib/spack/spack/test/util/lock_unix.py """ -import collections import errno -import getpass -import glob import multiprocessing -import os import pathlib -import shutil -import socket -import stat import sys -import tempfile -import traceback -from contextlib import contextmanager -from multiprocessing import Barrier, Process, Queue import pytest import spack.util.lock as lk +from spack.test.util.lock_backend import ( # noqa: F401 + lock_dir, + lock_fail_timeout, + lock_path, + read_only, +) from spack.util.filesystem import getuid, touch, working_dir +pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="fcntl is POSIX-only") + if sys.platform != "win32": import fcntl -pytestmark = pytest.mark.not_on_windows("does not run on windows") - - -# -# This test can be run with MPI. MPI is "enabled" if we can import -# mpi4py and the number of total MPI processes is greater than 1. -# Otherwise it just runs as a node-local test. -# -# NOTE: MPI mode is different from node-local mode in that node-local -# mode will spawn its own test processes, while MPI mode assumes you've -# run this script as a SPMD application. In MPI mode, no additional -# processes are spawned, and you need to ensure that you mpirun the -# script with enough processes for all the multiproc_test cases below. -# -# If you don't run with enough processes, tests that require more -# processes than you currently have will be skipped. -# -mpi = False -comm = None -try: - from mpi4py import MPI - - comm = MPI.COMM_WORLD - if comm.size > 1: - mpi = True -except ImportError: - pass - - -#: This is a list of filesystem locations to test locks in. Paths are -#: expanded so that %u is replaced with the current username. '~' is also -#: legal and will be expanded to the user's home directory. -#: -#: Tests are skipped for directories that don't exist, so you'll need to -#: update this with the locations of NFS, Lustre, and other mounts on your -#: system. -locations = [ - tempfile.gettempdir(), - os.path.join("/nfs/tmp2/", getpass.getuser()), - os.path.join("/p/lscratch*/", getpass.getuser()), -] - -#: This is the longest a failed multiproc test will take. -#: Barriers will time out and raise an exception after this interval. -#: In MPI mode, barriers don't time out (they hang). See mpi_multiproc_test. -barrier_timeout = 5 - -#: This is the lock timeout for expected failures. -#: This may need to be higher for some filesystems. -lock_fail_timeout = 0.1 - - -def make_readable(*paths): - # TODO: From os.chmod doc: - # "Note Although Windows supports chmod(), you can only - # set the file's read-only flag with it (via the stat.S_IWRITE and - # stat.S_IREAD constants or a corresponding integer value). All other - # bits are ignored." - for path in paths: - if sys.platform != "win32": - mode = 0o555 if os.path.isdir(path) else 0o444 - else: - mode = stat.S_IREAD - os.chmod(path, mode) - - -def make_writable(*paths): - for path in paths: - if sys.platform != "win32": - mode = 0o755 if os.path.isdir(path) else 0o744 - else: - mode = stat.S_IWRITE - os.chmod(path, mode) - - -@contextmanager -def read_only(*paths): - modes = [os.stat(p).st_mode for p in paths] - make_readable(*paths) - - yield - - for path, mode in zip(paths, modes): - os.chmod(path, mode) - - -@pytest.fixture(scope="session", params=locations) -def lock_test_directory(request): - """This fixture causes tests to be executed for many different mounts. - - See the ``locations`` dict above for details. - """ - return request.param - - -@pytest.fixture(scope="session") -def lock_dir(lock_test_directory): - parent = next( - (p for p in glob.glob(lock_test_directory) if os.path.exists(p) and os.access(p, os.W_OK)), - None, - ) - if not parent: - # Skip filesystems that don't exist or aren't writable - pytest.skip("requires filesystem: '%s'" % lock_test_directory) - elif mpi and parent == tempfile.gettempdir(): - # Skip local tmp test for MPI runs - pytest.skip("skipping local tmp directory for MPI test.") - - tempdir = None - if not mpi or comm.rank == 0: - tempdir = tempfile.mkdtemp(dir=parent) - if mpi: - tempdir = comm.bcast(tempdir) - - yield tempdir - - if mpi: - # rank 0 may get here before others, in which case it'll try to - # remove the directory while other processes try to re-create the - # lock. This will give errno 39: directory not empty. Use a - # barrier to ensure everyone is done first. - comm.barrier() - - if not mpi or comm.rank == 0: - make_writable(tempdir) - shutil.rmtree(tempdir) - - -@pytest.fixture -def private_lock_path(lock_dir): - """In MPI mode, this is a private lock for each rank in a multiproc test. - - For other modes, it is the same as a shared lock. - """ - lock_file = os.path.join(lock_dir, "lockfile") - if mpi: - lock_file += ".%s" % comm.rank - - yield lock_file - - if os.path.exists(lock_file): - make_writable(lock_dir, lock_file) - os.unlink(lock_file) - - -@pytest.fixture -def lock_path(lock_dir): - """This lock is shared among all processes in a multiproc test.""" - lock_file = os.path.join(lock_dir, "lockfile") - - yield lock_file - - if os.path.exists(lock_file): - make_writable(lock_dir, lock_file) - os.unlink(lock_file) - - -def test_poll_interval_generator(): - interval_iter = iter(lk.Lock._poll_interval_generator(_wait_times=[1, 2, 3])) - intervals = [next(interval_iter) for i in range(100)] - assert intervals == [1] * 20 + [2] * 40 + [3] * 40 - - -def local_multiproc_test(*functions, **kwargs): - """Order some processes using simple barrier synchronization.""" - b = Barrier(len(functions), timeout=barrier_timeout) - - args = (b,) + tuple(kwargs.get("extra_args", ())) - procs = [Process(target=f, args=args, name=f.__name__) for f in functions] - - for p in procs: - p.start() - - for p in procs: - p.join() - - assert all(p.exitcode == 0 for p in procs) - - -def mpi_multiproc_test(*functions): - """SPMD version of multiproc test. - - This needs to be run like so: - - srun spack test lock - - Each process executes its corresponding function. This is different - from ``multiproc_test`` above, which spawns the processes. This will - skip tests if there are too few processes to run them. - """ - procs = len(functions) - if procs > comm.size: - pytest.skip("requires at least %d MPI processes" % procs) - - comm.Barrier() # barrier before each MPI test - - include = comm.rank < len(functions) - subcomm = comm.Split(include) - - class subcomm_barrier: - """Stand-in for multiproc barrier for MPI-parallel jobs.""" - - def wait(self): - subcomm.Barrier() - - if include: - try: - functions[subcomm.rank](subcomm_barrier()) - except BaseException: - # aborting is the best we can do for MPI tests without - # hanging, since we're using MPI barriers. This will fail - # early and it loses the nice pytest output, but at least it - # gets use a stacktrace on the processes that failed. - traceback.print_exc() - comm.Abort() - subcomm.Free() - - comm.Barrier() # barrier after each MPI test. - - -#: ``multiproc_test()`` should be called by tests below. -#: ``multiproc_test()`` will work for either MPI runs or for local runs. -multiproc_test = mpi_multiproc_test if mpi else local_multiproc_test - - -# -# Process snippets below can be composed into tests. -# -class AcquireWrite: - def __init__(self, lock_path, start=0, length=0): - self.lock_path = lock_path - self.start = start - self.length = length - - @property - def __name__(self): - return self.__class__.__name__ - - def __call__(self, barrier): - lock = lk.Lock(self.lock_path, start=self.start, length=self.length) - lock.acquire_write() # grab exclusive lock - barrier.wait() - barrier.wait() # hold the lock until timeout in other procs. - - -class AcquireRead: - def __init__(self, lock_path, start=0, length=0): - self.lock_path = lock_path - self.start = start - self.length = length - - @property - def __name__(self): - return self.__class__.__name__ - - def __call__(self, barrier): - lock = lk.Lock(self.lock_path, start=self.start, length=self.length) - lock.acquire_read() # grab shared lock - barrier.wait() - barrier.wait() # hold the lock until timeout in other procs. - - -class TimeoutWrite: - def __init__(self, lock_path, start=0, length=0): - self.lock_path = lock_path - self.start = start - self.length = length - - @property - def __name__(self): - return self.__class__.__name__ - - def __call__(self, barrier): - lock = lk.Lock(self.lock_path, start=self.start, length=self.length) - barrier.wait() # wait for lock acquire in first process - with pytest.raises(lk.LockTimeoutError): - lock.acquire_write(lock_fail_timeout) - barrier.wait() - - -class TimeoutRead: - def __init__(self, lock_path, start=0, length=0): - self.lock_path = lock_path - self.start = start - self.length = length - - @property - def __name__(self): - return self.__class__.__name__ - - def __call__(self, barrier): - lock = lk.Lock(self.lock_path, start=self.start, length=self.length) - barrier.wait() # wait for lock acquire in first process - with pytest.raises(lk.LockTimeoutError): - lock.acquire_read(lock_fail_timeout) - barrier.wait() - - -# -# Test that exclusive locks on other processes time out when an -# exclusive lock is held. -# -def test_write_lock_timeout_on_write(lock_path): - multiproc_test(AcquireWrite(lock_path), TimeoutWrite(lock_path)) - - -def test_write_lock_timeout_on_write_2(lock_path): - multiproc_test(AcquireWrite(lock_path), TimeoutWrite(lock_path), TimeoutWrite(lock_path)) - - -def test_write_lock_timeout_on_write_3(lock_path): - multiproc_test( - AcquireWrite(lock_path), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - ) - - -def test_write_lock_timeout_on_write_ranges(lock_path): - multiproc_test(AcquireWrite(lock_path, 0, 1), TimeoutWrite(lock_path, 0, 1)) - - -def test_write_lock_timeout_on_write_ranges_2(lock_path): - multiproc_test( - AcquireWrite(lock_path, 0, 64), - AcquireWrite(lock_path, 65, 1), - TimeoutWrite(lock_path, 0, 1), - TimeoutWrite(lock_path, 63, 1), - ) - - -def test_write_lock_timeout_on_write_ranges_3(lock_path): - multiproc_test( - AcquireWrite(lock_path, 0, 1), - AcquireWrite(lock_path, 1, 1), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - ) - - -def test_write_lock_timeout_on_write_ranges_4(lock_path): - multiproc_test( - AcquireWrite(lock_path, 0, 1), - AcquireWrite(lock_path, 1, 1), - AcquireWrite(lock_path, 2, 456), - AcquireWrite(lock_path, 500, 64), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - ) - - -# -# Test that shared locks on other processes time out when an -# exclusive lock is held. -# -def test_read_lock_timeout_on_write(lock_path): - multiproc_test(AcquireWrite(lock_path), TimeoutRead(lock_path)) - - -def test_read_lock_timeout_on_write_2(lock_path): - multiproc_test(AcquireWrite(lock_path), TimeoutRead(lock_path), TimeoutRead(lock_path)) - - -def test_read_lock_timeout_on_write_3(lock_path): - multiproc_test( - AcquireWrite(lock_path), - TimeoutRead(lock_path), - TimeoutRead(lock_path), - TimeoutRead(lock_path), - ) - - -def test_read_lock_timeout_on_write_ranges(lock_path): - """small write lock, read whole file.""" - multiproc_test(AcquireWrite(lock_path, 0, 1), TimeoutRead(lock_path)) - - -def test_read_lock_timeout_on_write_ranges_2(lock_path): - """small write lock, small read lock""" - multiproc_test(AcquireWrite(lock_path, 0, 1), TimeoutRead(lock_path, 0, 1)) - - -def test_read_lock_timeout_on_write_ranges_3(lock_path): - """two write locks, overlapping read locks""" - multiproc_test( - AcquireWrite(lock_path, 0, 1), - AcquireWrite(lock_path, 64, 128), - TimeoutRead(lock_path, 0, 1), - TimeoutRead(lock_path, 128, 256), - ) - - -# -# Test that exclusive locks time out when shared locks are held. -# -def test_write_lock_timeout_on_read(lock_path): - multiproc_test(AcquireRead(lock_path), TimeoutWrite(lock_path)) - - -def test_write_lock_timeout_on_read_2(lock_path): - multiproc_test(AcquireRead(lock_path), TimeoutWrite(lock_path), TimeoutWrite(lock_path)) - - -def test_write_lock_timeout_on_read_3(lock_path): - multiproc_test( - AcquireRead(lock_path), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - ) - - -def test_write_lock_timeout_on_read_ranges(lock_path): - multiproc_test(AcquireRead(lock_path, 0, 1), TimeoutWrite(lock_path)) - - -def test_write_lock_timeout_on_read_ranges_2(lock_path): - multiproc_test(AcquireRead(lock_path, 0, 1), TimeoutWrite(lock_path, 0, 1)) - - -def test_write_lock_timeout_on_read_ranges_3(lock_path): - multiproc_test( - AcquireRead(lock_path, 0, 1), - AcquireRead(lock_path, 10, 1), - TimeoutWrite(lock_path, 0, 1), - TimeoutWrite(lock_path, 10, 1), - ) - - -def test_write_lock_timeout_on_read_ranges_4(lock_path): - multiproc_test( - AcquireRead(lock_path, 0, 64), - TimeoutWrite(lock_path, 10, 1), - TimeoutWrite(lock_path, 32, 1), - ) - - -def test_write_lock_timeout_on_read_ranges_5(lock_path): - multiproc_test( - AcquireRead(lock_path, 64, 128), - TimeoutWrite(lock_path, 65, 1), - TimeoutWrite(lock_path, 127, 1), - TimeoutWrite(lock_path, 90, 10), - ) - - -# -# Test that exclusive locks time while lots of shared locks are held. -# -def test_write_lock_timeout_with_multiple_readers_2_1(lock_path): - multiproc_test(AcquireRead(lock_path), AcquireRead(lock_path), TimeoutWrite(lock_path)) - - -def test_write_lock_timeout_with_multiple_readers_2_2(lock_path): - multiproc_test( - AcquireRead(lock_path), - AcquireRead(lock_path), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - ) - - -def test_write_lock_timeout_with_multiple_readers_3_1(lock_path): - multiproc_test( - AcquireRead(lock_path), - AcquireRead(lock_path), - AcquireRead(lock_path), - TimeoutWrite(lock_path), - ) - - -def test_write_lock_timeout_with_multiple_readers_3_2(lock_path): - multiproc_test( - AcquireRead(lock_path), - AcquireRead(lock_path), - AcquireRead(lock_path), - TimeoutWrite(lock_path), - TimeoutWrite(lock_path), - ) - - -def test_write_lock_timeout_with_multiple_readers_2_1_ranges(lock_path): - multiproc_test( - AcquireRead(lock_path, 0, 10), AcquireRead(lock_path, 2, 10), TimeoutWrite(lock_path, 5, 5) - ) - - -def test_write_lock_timeout_with_multiple_readers_2_3_ranges(lock_path): - multiproc_test( - AcquireRead(lock_path, 0, 10), - AcquireRead(lock_path, 5, 15), - TimeoutWrite(lock_path, 0, 1), - TimeoutWrite(lock_path, 11, 3), - TimeoutWrite(lock_path, 7, 1), - ) - - -def test_write_lock_timeout_with_multiple_readers_3_1_ranges(lock_path): - multiproc_test( - AcquireRead(lock_path, 0, 5), - AcquireRead(lock_path, 5, 5), - AcquireRead(lock_path, 10, 5), - TimeoutWrite(lock_path, 0, 15), - ) - - -def test_write_lock_timeout_with_multiple_readers_3_2_ranges(lock_path): - multiproc_test( - AcquireRead(lock_path, 0, 5), - AcquireRead(lock_path, 5, 5), - AcquireRead(lock_path, 10, 5), - TimeoutWrite(lock_path, 3, 10), - TimeoutWrite(lock_path, 5, 1), - ) - @pytest.mark.skipif(getuid() == 0, reason="user is root") def test_read_lock_on_read_only_lockfile(lock_dir, lock_path): @@ -584,20 +47,7 @@ def test_read_lock_on_read_only_lockfile(lock_dir, lock_path): pass -def test_read_lock_read_only_dir_writable_lockfile(lock_dir, lock_path): - """read-only directory, writable lockfile.""" - touch(lock_path) - with read_only(lock_dir): - lock = lk.Lock(lock_path) - - with lk.ReadTransaction(lock): - pass - - with lk.WriteTransaction(lock): - pass - - -@pytest.mark.skipif(False if sys.platform == "win32" else getuid() == 0, reason="user is root") +@pytest.mark.skipif(getuid() == 0, reason="user is root") def test_read_lock_no_lockfile(lock_dir, lock_path): """read-only directory, no lockfile (so can't create).""" with read_only(lock_dir): @@ -612,720 +62,6 @@ def test_read_lock_no_lockfile(lock_dir, lock_path): pass -def test_upgrade_read_to_write(private_lock_path): - """Test that a read lock can be upgraded to a write lock. - - Note that to upgrade a read lock to a write lock, you have the be the - only holder of a read lock. Client code needs to coordinate that for - shared locks. For this test, we use a private lock just to test that an - upgrade is possible. - """ - # ensure lock file exists the first time, so we open it read-only - # to begin with. - touch(private_lock_path) - - lock = lk.Lock(private_lock_path) - assert lock._reads == 0 - assert lock._writes == 0 - - lock.acquire_read() - assert lock._reads == 1 - assert lock._writes == 0 - assert lock.backend._file_ref.fh.mode == "rb+" - - lock.acquire_write() - assert lock._reads == 1 - assert lock._writes == 1 - assert lock.backend._file_ref.fh.mode == "rb+" - - lock.release_write() - assert lock._reads == 1 - assert lock._writes == 0 - assert lock.backend._file_ref.fh.mode == "rb+" - - lock.release_read() - assert lock._reads == 0 - assert lock._writes == 0 - assert not lock.backend._file_ref.fh.closed # recycle the file handle for next lock - - -def test_release_write_downgrades_to_shared(private_lock_path): - """Releasing a write lock while a read lock is held must downgrade the POSIX lock - from exclusive to shared, allowing other processes to acquire read locks.""" - lock = lk.Lock(private_lock_path) - lock.acquire_read() - lock.acquire_write() - lock.release_write() - assert lock._reads == 1 - assert lock._writes == 0 - - ctx = multiprocessing.get_context() - q = ctx.Queue() - - # Another process must be able to acquire a shared read lock concurrently. - p = ctx.Process(target=_child_try_acquire_read, args=(private_lock_path, q)) - p.start() - p.join() - assert q.get() is True - - # But must not be able to acquire an exclusive write lock. - p = ctx.Process(target=_child_try_acquire_write, args=(private_lock_path, q)) - p.start() - p.join() - assert q.get() is False - - lock.release_read() - assert lock._reads == 0 - assert lock._writes == 0 - - -@pytest.mark.skipif(getuid() == 0, reason="user is root") -def test_upgrade_read_to_write_fails_with_readonly_file(private_lock_path): - """Test that read-only file can be read-locked but not write-locked.""" - # ensure lock file exists the first time - touch(private_lock_path) - - # open it read-only to begin with. - with read_only(private_lock_path): - lock = lk.Lock(private_lock_path) - assert lock._reads == 0 - assert lock._writes == 0 - - lock.acquire_read() - assert lock._reads == 1 - assert lock._writes == 0 - assert lock.backend._file_ref.fh.mode == "rb" - - # upgrade to write here - with pytest.raises(lk.LockROFileError): - lock.acquire_write() - - -class ComplexAcquireAndRelease: - def __init__(self, lock_path): - self.lock_path = lock_path - - def p1(self, barrier): - lock = lk.Lock(self.lock_path) - - lock.acquire_write() - barrier.wait() # ---------------------------------------- 1 - # others test timeout - barrier.wait() # ---------------------------------------- 2 - lock.release_write() # release and others acquire read - barrier.wait() # ---------------------------------------- 3 - with pytest.raises(lk.LockTimeoutError): - lock.acquire_write(lock_fail_timeout) - lock.acquire_read() - barrier.wait() # ---------------------------------------- 4 - lock.release_read() - barrier.wait() # ---------------------------------------- 5 - - # p2 upgrades read to write - barrier.wait() # ---------------------------------------- 6 - with pytest.raises(lk.LockTimeoutError): - lock.acquire_write(lock_fail_timeout) - with pytest.raises(lk.LockTimeoutError): - lock.acquire_read(lock_fail_timeout) - barrier.wait() # ---------------------------------------- 7 - # p2 releases write and read - barrier.wait() # ---------------------------------------- 8 - - # p3 acquires read - barrier.wait() # ---------------------------------------- 9 - # p3 upgrades read to write - barrier.wait() # ---------------------------------------- 10 - with pytest.raises(lk.LockTimeoutError): - lock.acquire_write(lock_fail_timeout) - with pytest.raises(lk.LockTimeoutError): - lock.acquire_read(lock_fail_timeout) - barrier.wait() # ---------------------------------------- 11 - # p3 releases locks - barrier.wait() # ---------------------------------------- 12 - lock.acquire_read() - barrier.wait() # ---------------------------------------- 13 - lock.release_read() - - def p2(self, barrier): - lock = lk.Lock(self.lock_path) - - # p1 acquires write - barrier.wait() # ---------------------------------------- 1 - with pytest.raises(lk.LockTimeoutError): - lock.acquire_write(lock_fail_timeout) - with pytest.raises(lk.LockTimeoutError): - lock.acquire_read(lock_fail_timeout) - barrier.wait() # ---------------------------------------- 2 - lock.acquire_read() - barrier.wait() # ---------------------------------------- 3 - # p1 tests shared read - barrier.wait() # ---------------------------------------- 4 - # others release reads - barrier.wait() # ---------------------------------------- 5 - - lock.acquire_write() # upgrade read to write - barrier.wait() # ---------------------------------------- 6 - # others test timeout - barrier.wait() # ---------------------------------------- 7 - lock.release_write() # release read AND write (need both) - lock.release_read() - barrier.wait() # ---------------------------------------- 8 - - # p3 acquires read - barrier.wait() # ---------------------------------------- 9 - # p3 upgrades read to write - barrier.wait() # ---------------------------------------- 10 - with pytest.raises(lk.LockTimeoutError): - lock.acquire_write(lock_fail_timeout) - with pytest.raises(lk.LockTimeoutError): - lock.acquire_read(lock_fail_timeout) - barrier.wait() # ---------------------------------------- 11 - # p3 releases locks - barrier.wait() # ---------------------------------------- 12 - lock.acquire_read() - barrier.wait() # ---------------------------------------- 13 - lock.release_read() - - def p3(self, barrier): - lock = lk.Lock(self.lock_path) - - # p1 acquires write - barrier.wait() # ---------------------------------------- 1 - with pytest.raises(lk.LockTimeoutError): - lock.acquire_write(lock_fail_timeout) - with pytest.raises(lk.LockTimeoutError): - lock.acquire_read(lock_fail_timeout) - barrier.wait() # ---------------------------------------- 2 - lock.acquire_read() - barrier.wait() # ---------------------------------------- 3 - # p1 tests shared read - barrier.wait() # ---------------------------------------- 4 - lock.release_read() - barrier.wait() # ---------------------------------------- 5 - - # p2 upgrades read to write - barrier.wait() # ---------------------------------------- 6 - with pytest.raises(lk.LockTimeoutError): - lock.acquire_write(lock_fail_timeout) - with pytest.raises(lk.LockTimeoutError): - lock.acquire_read(lock_fail_timeout) - barrier.wait() # ---------------------------------------- 7 - # p2 releases write & read - barrier.wait() # ---------------------------------------- 8 - - lock.acquire_read() - barrier.wait() # ---------------------------------------- 9 - lock.acquire_write() - barrier.wait() # ---------------------------------------- 10 - # others test timeout - barrier.wait() # ---------------------------------------- 11 - lock.release_read() # release read AND write in opposite - lock.release_write() # order from before on p2 - barrier.wait() # ---------------------------------------- 12 - lock.acquire_read() - barrier.wait() # ---------------------------------------- 13 - lock.release_read() - - -# -# Longer test case that ensures locks are reusable. Ordering is -# enforced by barriers throughout -- steps are shown with numbers. -# -def test_complex_acquire_and_release_chain(lock_path): - test_chain = ComplexAcquireAndRelease(lock_path) - multiproc_test(test_chain.p1, test_chain.p2, test_chain.p3) - - -class AssertLock(lk.Lock): - """Test lock class that marks acquire/release events.""" - - def __init__(self, lock_path, vals): - super().__init__(lock_path) - self.vals = vals - - # assert hooks for subclasses - assert_acquire_read = lambda self: None - assert_acquire_write = lambda self: None - assert_release_read = lambda self: None - assert_release_write = lambda self: None - - def acquire_read(self, timeout=None): - self.assert_acquire_read() - result = super().acquire_read(timeout) - self.vals["acquired_read"] = True - return result - - def acquire_write(self, timeout=None): - self.assert_acquire_write() - result = super().acquire_write(timeout) - self.vals["acquired_write"] = True - return result - - def release_read(self, release_fn=None): - self.assert_release_read() - result = super().release_read(release_fn) - self.vals["released_read"] = True - return result - - def release_write(self, release_fn=None): - self.assert_release_write() - result = super().release_write(release_fn) - self.vals["released_write"] = True - return result - - -@pytest.mark.parametrize( - "transaction,type", [(lk.ReadTransaction, "read"), (lk.WriteTransaction, "write")] -) -def test_transaction(lock_path, transaction, type): - class MockLock(AssertLock): - def assert_acquire_read(self): - assert not vals["entered_fn"] - assert not vals["exited_fn"] - - def assert_release_read(self): - assert vals["entered_fn"] - assert not vals["exited_fn"] - - def assert_acquire_write(self): - assert not vals["entered_fn"] - assert not vals["exited_fn"] - - def assert_release_write(self): - assert vals["entered_fn"] - assert not vals["exited_fn"] - - def enter_fn(): - # assert enter_fn is called while lock is held - assert vals["acquired_%s" % type] - vals["entered_fn"] = True - - def exit_fn(t, v, tb): - # assert exit_fn is called while lock is held - assert not vals["released_%s" % type] - vals["exited_fn"] = True - vals["exception"] = t or v or tb - - vals = collections.defaultdict(lambda: False) - lock = MockLock(lock_path, vals) - - with transaction(lock, acquire=enter_fn, release=exit_fn): - assert vals["acquired_%s" % type] - assert not vals["released_%s" % type] - - assert vals["entered_fn"] - assert vals["exited_fn"] - assert vals["acquired_%s" % type] - assert vals["released_%s" % type] - assert not vals["exception"] - - -@pytest.mark.parametrize( - "transaction,type", [(lk.ReadTransaction, "read"), (lk.WriteTransaction, "write")] -) -def test_transaction_with_exception(lock_path, transaction, type): - class MockLock(AssertLock): - def assert_acquire_read(self): - assert not vals["entered_fn"] - assert not vals["exited_fn"] - - def assert_release_read(self): - assert vals["entered_fn"] - assert not vals["exited_fn"] - - def assert_acquire_write(self): - assert not vals["entered_fn"] - assert not vals["exited_fn"] - - def assert_release_write(self): - assert vals["entered_fn"] - assert not vals["exited_fn"] - - def enter_fn(): - assert vals["acquired_%s" % type] - vals["entered_fn"] = True - - def exit_fn(t, v, tb): - assert not vals["released_%s" % type] - vals["exited_fn"] = True - vals["exception"] = t or v or tb - return exit_result - - exit_result = False - vals = collections.defaultdict(lambda: False) - lock = MockLock(lock_path, vals) - - with pytest.raises(Exception): - with transaction(lock, acquire=enter_fn, release=exit_fn): - raise Exception() - - assert vals["entered_fn"] - assert vals["exited_fn"] - assert vals["exception"] - - # test suppression of exceptions from exit_fn - exit_result = True - vals.clear() - - # should not raise now. - with transaction(lock, acquire=enter_fn, release=exit_fn): - raise Exception() - - assert vals["entered_fn"] - assert vals["exited_fn"] - assert vals["exception"] - - -def test_nested_write_transaction(lock_path): - """Ensure that the outermost write transaction writes.""" - - def write(t, v, tb): - vals["wrote"] = True - - vals = collections.defaultdict(lambda: False) - lock = AssertLock(lock_path, vals) - - # write/write - with lk.WriteTransaction(lock, release=write): - assert not vals["wrote"] - with lk.WriteTransaction(lock, release=write): - assert not vals["wrote"] - assert not vals["wrote"] - assert vals["wrote"] - - # read/write - vals.clear() - with lk.ReadTransaction(lock): - assert not vals["wrote"] - with lk.WriteTransaction(lock, release=write): - assert not vals["wrote"] - assert vals["wrote"] - - # write/read/write - vals.clear() - with lk.WriteTransaction(lock, release=write): - assert not vals["wrote"] - with lk.ReadTransaction(lock): - assert not vals["wrote"] - with lk.WriteTransaction(lock, release=write): - assert not vals["wrote"] - assert not vals["wrote"] - assert not vals["wrote"] - assert vals["wrote"] - - # read/write/read/write - vals.clear() - with lk.ReadTransaction(lock): - with lk.WriteTransaction(lock, release=write): - assert not vals["wrote"] - with lk.ReadTransaction(lock): - assert not vals["wrote"] - with lk.WriteTransaction(lock, release=write): - assert not vals["wrote"] - assert not vals["wrote"] - assert not vals["wrote"] - assert vals["wrote"] - - -def test_nested_reads(lock_path): - """Ensure that write transactions won't re-read data.""" - - def read(): - vals["read"] += 1 - - vals = collections.defaultdict(lambda: 0) - lock = AssertLock(lock_path, vals) - - # read/read - vals.clear() - assert vals["read"] == 0 - with lk.ReadTransaction(lock, acquire=read): - assert vals["read"] == 1 - with lk.ReadTransaction(lock, acquire=read): - assert vals["read"] == 1 - - # write/write - vals.clear() - assert vals["read"] == 0 - with lk.WriteTransaction(lock, acquire=read): - assert vals["read"] == 1 - with lk.WriteTransaction(lock, acquire=read): - assert vals["read"] == 1 - - # read/write - vals.clear() - assert vals["read"] == 0 - with lk.ReadTransaction(lock, acquire=read): - assert vals["read"] == 1 - with lk.WriteTransaction(lock, acquire=read): - assert vals["read"] == 1 - - # write/read/write - vals.clear() - assert vals["read"] == 0 - with lk.WriteTransaction(lock, acquire=read): - assert vals["read"] == 1 - with lk.ReadTransaction(lock, acquire=read): - assert vals["read"] == 1 - with lk.WriteTransaction(lock, acquire=read): - assert vals["read"] == 1 - - # read/write/read/write - vals.clear() - assert vals["read"] == 0 - with lk.ReadTransaction(lock, acquire=read): - assert vals["read"] == 1 - with lk.WriteTransaction(lock, acquire=read): - assert vals["read"] == 1 - with lk.ReadTransaction(lock, acquire=read): - assert vals["read"] == 1 - with lk.WriteTransaction(lock, acquire=read): - assert vals["read"] == 1 - - -@pytest.mark.parametrize( - "transaction,type", [(lk.TryReadTransaction, "read"), (lk.TryWriteTransaction, "write")] -) -def test_try_transaction(lock_path, transaction, type): - """When uncontended, try-transactions yield True and behave like blocking transactions.""" - - def enter_fn(): - vals["entered_fn"] = True - - def exit_fn(t, v, tb): - vals["exited_fn"] = True - vals["exception"] = t or v or tb - - vals = collections.defaultdict(lambda: False) - lock = AssertLock(lock_path, vals) - - with transaction(lock, acquire=enter_fn, release=exit_fn) as acquired: - assert acquired - assert vals["entered_fn"] - assert not vals["released_%s" % type] - - assert vals["exited_fn"] - assert vals["released_%s" % type] - assert not vals["exception"] - - -@pytest.mark.parametrize( - "transaction,attr", - [(lk.TryReadTransaction, "try_acquire_read"), (lk.TryWriteTransaction, "try_acquire_write")], -) -def test_try_transaction_blocked(lock_path, transaction, attr, monkeypatch): - """When the lock would block, try-transactions yield False and call no functions on exit.""" - - def enter_fn(): - vals["entered_fn"] = True - - def exit_fn(t, v, tb): - vals["exited_fn"] = True - - vals = collections.defaultdict(lambda: False) - lock = AssertLock(lock_path, vals) - monkeypatch.setattr(lock, attr, lambda: False) - - with transaction(lock, acquire=enter_fn, release=exit_fn) as acquired: - assert not acquired - - assert not vals["entered_fn"] - assert not vals["exited_fn"] - assert lock._reads == 0 and lock._writes == 0 - - # exceptions raised in the body still propagate - with pytest.raises(ValueError): - with transaction(lock, acquire=enter_fn, release=exit_fn) as acquired: - assert not acquired - raise ValueError() - - -def test_try_transaction_with_exception(lock_path): - """An exception in the body propagates and is forwarded to the release function.""" - - def exit_fn(t, v, tb): - vals["exception"] = t or v or tb - - vals = collections.defaultdict(lambda: False) - lock = AssertLock(lock_path, vals) - - with pytest.raises(ValueError): - with lk.TryWriteTransaction(lock, release=exit_fn) as acquired: - assert acquired - raise ValueError() - - assert vals["exception"] - assert vals["released_write"] - assert lock._reads == 0 and lock._writes == 0 - - -def test_try_transaction_nested(lock_path): - """Nested try-transactions acquire, but do not re-run the acquire function, and the release - function only runs when the outermost write lock is released.""" - - def read(): - vals["read"] += 1 - - def write(t, v, tb): - vals["wrote"] += 1 - - vals = collections.defaultdict(lambda: 0) - lock = AssertLock(lock_path, vals) - - with lk.WriteTransaction(lock, acquire=read, release=write): - assert vals["read"] == 1 - with lk.TryWriteTransaction(lock, acquire=read, release=write) as acquired: - assert acquired - assert vals["read"] == 1 - assert vals["wrote"] == 0 - with lk.TryReadTransaction(lock, acquire=read) as acquired: - assert acquired - assert vals["read"] == 1 - assert lock._writes == 1 - assert vals["wrote"] == 1 - assert lock._reads == 0 and lock._writes == 0 - - -class LockDebugOutput: - def __init__(self, lock_path): - self.lock_path = lock_path - self.host = socket.gethostname() - - def p1(self, barrier, q1, q2): - # exchange pids - p1_pid = os.getpid() - q1.put(p1_pid) - p2_pid = q2.get() - - # set up lock - lock = lk.Lock(self.lock_path, debug=True) - - with lk.WriteTransaction(lock): - # p1 takes write lock and writes pid/host to file - barrier.wait() # ------------------------------------ 1 - - assert lock.backend.pid == p1_pid - assert lock.backend.host == self.host - - # wait for p2 to verify contents of file - barrier.wait() # ---------------------------------------- 2 - - # wait for p2 to take a write lock - barrier.wait() # ---------------------------------------- 3 - - # verify pid/host info again - with lk.ReadTransaction(lock): - assert lock.backend.old_pid == p1_pid - assert lock.backend.old_host == self.host - - assert lock.backend.pid == p2_pid - assert lock.backend.host == self.host - - barrier.wait() # ---------------------------------------- 4 - - def p2(self, barrier, q1, q2): - # exchange pids - p2_pid = os.getpid() - p1_pid = q1.get() - q2.put(p2_pid) - - # set up lock - lock = lk.Lock(self.lock_path, debug=True) - - # p1 takes write lock and writes pid/host to file - barrier.wait() # ---------------------------------------- 1 - - # verify that p1 wrote information to lock file - with lk.ReadTransaction(lock): - assert lock.backend.pid == p1_pid - assert lock.backend.host == self.host - - barrier.wait() # ---------------------------------------- 2 - - # take a write lock on the file and verify pid/host info - with lk.WriteTransaction(lock): - assert lock.backend.old_pid == p1_pid - assert lock.backend.old_host == self.host - - assert lock.backend.pid == p2_pid - assert lock.backend.host == self.host - - barrier.wait() # ------------------------------------ 3 - - # wait for p1 to verify pid/host info - barrier.wait() # ---------------------------------------- 4 - - -def test_lock_debug_output(lock_path): - test_debug = LockDebugOutput(lock_path) - q1, q2 = Queue(), Queue() - local_multiproc_test(test_debug.p2, test_debug.p1, extra_args=(q1, q2)) - - -def test_lock_with_no_parent_directory(tmp_path: pathlib.Path): - """Make sure locks work even when their parent directory does not exist.""" - with working_dir(str(tmp_path)): - lock = lk.Lock("foo/bar/baz/lockfile") - with lk.WriteTransaction(lock): - pass - - -def test_lock_in_current_directory(tmp_path: pathlib.Path): - """Make sure locks work even when their parent directory does not exist.""" - with working_dir(str(tmp_path)): - # test we can create a lock in the current directory - lock = lk.Lock("lockfile") - for i in range(10): - with lk.ReadTransaction(lock): - pass - with lk.WriteTransaction(lock): - pass - - # and that we can do the same thing after it's already there - lock = lk.Lock("lockfile") - for i in range(10): - with lk.ReadTransaction(lock): - pass - with lk.WriteTransaction(lock): - pass - - -def test_attempts_str(): - assert lk._attempts_str(0, 0) == "" - assert lk._attempts_str(0.12, 1) == "" - assert lk._attempts_str(12.345, 2) == " after 12.345s and 2 attempts" - - -def test_lock_str(): - lock = lk.Lock("lockfile") - lockstr = str(lock) - assert "lockfile[0:0]" in lockstr - assert "timeout=None" in lockstr - assert "#reads=0, #writes=0" in lockstr - - -def test_downgrade_write_okay(tmp_path: pathlib.Path): - """Test the lock write-to-read downgrade operation.""" - with working_dir(str(tmp_path)): - lock = lk.Lock("lockfile") - lock.acquire_write() - lock.downgrade_write_to_read() - assert lock._reads == 1 - assert lock._writes == 0 - lock.release_read() - - -def test_downgrade_write_fails(tmp_path: pathlib.Path): - """Test failing the lock write-to-read downgrade operation.""" - with working_dir(str(tmp_path)): - lock = lk.Lock("lockfile") - lock.acquire_read() - msg = "Cannot downgrade lock from write to read on file: lockfile" - with pytest.raises(lk.LockDowngradeError, match=msg): - lock.downgrade_write_to_read() - lock.release_read() - - @pytest.mark.parametrize( "err_num,err_msg", [ @@ -1335,7 +71,10 @@ def test_downgrade_write_fails(tmp_path: pathlib.Path): ], ) def test_poll_lock_exception(tmp_path: pathlib.Path, monkeypatch, err_num, err_msg): - """Test poll lock exception handling.""" + """A ``fcntl.lockf`` failure with EACCES/EAGAIN means someone else holds the lock: poll() + should report that as a failed (non-blocking) attempt, not raise. Any other errno is a real + error and should propagate. + """ def _lockf(fd, cmd, len, start, whence): raise OSError(err_num, err_msg) @@ -1348,37 +87,15 @@ def _lockf(fd, cmd, len, start, whence): monkeypatch.setattr(fcntl, "lockf", _lockf) if err_num in [errno.EAGAIN, errno.EACCES]: - assert not lock.backend.poll(fcntl.LOCK_EX) + assert not lock._poll_lock(lk.LockType.LOCK_EX) else: with pytest.raises(OSError, match=err_msg): - lock.backend.poll(fcntl.LOCK_EX) + lock._poll_lock(lk.LockType.LOCK_EX) monkeypatch.undo() lock.release_read() -def test_upgrade_read_okay(tmp_path: pathlib.Path): - """Test the lock read-to-write upgrade operation.""" - with working_dir(str(tmp_path)): - lock = lk.Lock("lockfile") - lock.acquire_read() - lock.upgrade_read_to_write() - assert lock._reads == 0 - assert lock._writes == 1 - lock.release_write() - - -def test_upgrade_read_fails(tmp_path: pathlib.Path): - """Test failing the lock read-to-write upgrade operation.""" - with working_dir(str(tmp_path)): - lock = lk.Lock("lockfile") - lock.acquire_write() - msg = "Cannot upgrade lock from read to write on file: lockfile" - with pytest.raises(lk.LockUpgradeError, match=msg): - lock.upgrade_read_to_write() - lock.release_write() - - @pytest.mark.parametrize("acquire", ["acquire_write", "acquire_read"]) def test_acquire_after_fork(tmp_path: pathlib.Path, acquire: str): """After fork, acquire_write/read must not silently succeed due to inherited counters.""" @@ -1412,105 +129,3 @@ def child(): assert result.get() == "timed_out" finally: lock.release_write() - - -def _child_try_acquire_write(lock_path: str, result_queue): - lock = lk.Lock(lock_path) - result_queue.put(lock.try_acquire_write()) - - -def _child_try_acquire_read(lock_path: str, result_queue): - lock = lk.Lock(lock_path) - result_queue.put(lock.try_acquire_read()) - - -def test_try_acquire_read(tmp_path: pathlib.Path): - """Test non-blocking try_acquire_read.""" - lock = lk.Lock(str(tmp_path / "lockfile")) - - # Succeeds on unlocked lock - assert lock.try_acquire_read() is True - assert lock._reads == 1 - - # Succeeds again (nested) - assert lock.try_acquire_read() is True - assert lock._reads == 2 - - lock.release_read() - lock.release_read() - ctx = multiprocessing.get_context() - - # Fails when another process holds an exclusive write lock - lock.acquire_write() - try: - q = ctx.Queue() - p = ctx.Process(target=_child_try_acquire_read, args=(str(tmp_path / "lockfile"), q)) - p.start() - p.join() - assert q.get() is False - finally: - lock.release_write() - - -def test_try_acquire_write(tmp_path: pathlib.Path): - """Test non-blocking try_acquire_write.""" - lock = lk.Lock(str(tmp_path / "lockfile")) - ctx = multiprocessing.get_context() - - # Succeeds on unlocked lock - assert lock.try_acquire_write() is True - assert lock._writes == 1 - - # Succeeds again (nested) - assert lock.try_acquire_write() is True - assert lock._writes == 2 - - lock.release_write() - lock.release_write() - - # Fails when another process holds a write lock - lock.acquire_write() - try: - q = ctx.Queue() - p = ctx.Process(target=_child_try_acquire_write, args=(str(tmp_path / "lockfile"), q)) - p.start() - p.join() - assert q.get() is False - finally: - lock.release_write() - - # Fails when another process holds a read lock - lock.acquire_read() - try: - q = ctx.Queue() - p = ctx.Process(target=_child_try_acquire_write, args=(str(tmp_path / "lockfile"), q)) - p.start() - p.join() - assert q.get() is False - finally: - lock.release_read() - - -def _child_fails_to_acquire_read(_lock: lk.Lock): - try: - _lock.acquire_read(timeout=1e-9) - except lk.LockTimeoutError: - return - assert False, "Child process should not have been able to acquire read lock" - - -def test_read_after_write_does_not_accidentally_downgrade(tmp_path: pathlib.Path): - """Test that acquiring a read lock after a write lock does not accidentally downgrade the - write lock, by having another process attempt to acquire a read lock.""" - lock = lk.Lock(str(tmp_path / "lockfile")) - lock.acquire_write() - lock.acquire_read() # should not downgrade the write lock - try: - # No matter the start method, the child process shouldn't be able to acquire a read lock. - p = multiprocessing.Process(target=_child_fails_to_acquire_read, args=(lock,)) - p.start() - p.join() - assert p.exitcode == 0 - finally: - lock.release_read() - lock.release_write() diff --git a/lib/spack/spack/test/util/lock_windows.py b/lib/spack/spack/test/util/lock_windows.py new file mode 100644 index 00000000000000..23b3e2eb0dd802 --- /dev/null +++ b/lib/spack/spack/test/util/lock_windows.py @@ -0,0 +1,66 @@ +# Copyright Spack Project Developers. See COPYRIGHT file for details. +# +# SPDX-License-Identifier: (Apache-2.0 OR MIT) + +"""Tests for the ctypes-based Windows locking internals in spack.util.lock_windows. + +These exercise behavior specific to WindowsBackend's LockFileEx error handling (see +spack.util.lock_windows._win_lock_file_ex): errors that mean "someone else holds this lock" must +be reported as a failed poll, not raised, while any other error is a real failure. +""" + +import pathlib +import sys + +import pytest + +if sys.platform != "win32": + pytest.skip("Windows-only tests", allow_module_level=True) + +import ctypes + +import spack.util.lock as lk +import spack.util.lock_windows as lkw +from spack.util.filesystem import working_dir + + +@pytest.mark.parametrize("winerror", [32, 33]) # ERROR_SHARING_VIOLATION, ERROR_LOCK_VIOLATION +def test_poll_lock_contended_win(tmp_path: pathlib.Path, monkeypatch, winerror): + """A LockFileEx failure with ERROR_SHARING_VIOLATION/ERROR_LOCK_VIOLATION means someone else + holds the lock: poll() should report that as a failed (non-blocking) attempt, not raise.""" + + def fake_lock_file_ex(handle, flags, reserved, low, high, overlapped): + ctypes.set_last_error(winerror) + return 0 + + with working_dir(str(tmp_path)): + lock = lk.Lock("lockfile") + lock.acquire_read() + + kernel32 = lkw._kernel32 # type: ignore[has-type] + monkeypatch.setattr(kernel32, "LockFileEx", fake_lock_file_ex) + assert not lock._poll_lock(lk.LockType.LOCK_EX) + monkeypatch.undo() + + lock.release_read() + + +def test_poll_lock_unexpected_error_win(tmp_path: pathlib.Path, monkeypatch): + """A LockFileEx failure with any winerror other than ERROR_SHARING_VIOLATION/ + ERROR_LOCK_VIOLATION is a real error, and should propagate as an OSError.""" + + def fake_lock_file_ex(handle, flags, reserved, low, high, overlapped): + ctypes.set_last_error(5) # ERROR_ACCESS_DENIED: not a contention code + return 0 + + with working_dir(str(tmp_path)): + lock = lk.Lock("lockfile") + lock.acquire_read() + + kernel32 = lkw._kernel32 # type: ignore[has-type] + monkeypatch.setattr(kernel32, "LockFileEx", fake_lock_file_ex) + with pytest.raises(OSError): + lock._poll_lock(lk.LockType.LOCK_EX) + monkeypatch.undo() + + lock.release_read() diff --git a/lib/spack/spack/util/lock.py b/lib/spack/spack/util/lock.py index 09b43fb7acd560..2492bff7280c8c 100644 --- a/lib/spack/spack/util/lock.py +++ b/lib/spack/spack/util/lock.py @@ -2,50 +2,57 @@ # # SPDX-License-Identifier: (Apache-2.0 OR MIT) -import errno import os -import socket import stat import sys import time from datetime import datetime from types import TracebackType -from typing import IO, Callable, Dict, Generator, Optional, Tuple, Type, Union +from typing import Callable, Generator, Optional, Tuple, Type # novm import spack.error from spack.util import lang, tty +from spack.util.lock_common import ( + FILE_TRACKER, + CantCreateLockError, + GenericLockBackend, + LockError, + LockPermissionError, + LockROFileError, +) from spack.util.string import plural -if sys.platform != "win32": - import fcntl + +if sys.platform == "win32": + from spack.util.lock_windows import LockType, WindowsBackend +else: + from spack.util.lock_posix import LockType, PosixBackend __all__ = [ - "check_lock_safety", - "CantCreateLockError", - "DummyBackend", "Lock", "LockDowngradeError", + "LockUpgradeError", + "LockTransaction", + "WriteTransaction", + "ReadTransaction", "LockError", + "LockTimeoutError", "LockPermissionError", "LockROFileError", - "LockTimeoutError", - "LockTransaction", - "LockUpgradeError", - "PosixBackend", - "ReadTransaction", - "TryReadTransaction", - "TryWriteTransaction", - "WriteTransaction", + "CantCreateLockError", + "DummyBackend", + "FILE_TRACKER", ] +WHOLE_FILE_RANGE = 0xFFFFFFFF if sys.platform == "win32" else 0 + ExitFnType = Callable[ [Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]], Optional[bool], ] ReleaseFnType = Optional[Callable[[], Optional[bool]]] -DevIno = Tuple[int, int] # (st_dev, st_ino) from os.stat_result def true_fn() -> bool: @@ -53,100 +60,6 @@ def true_fn() -> bool: return True -class OpenFile: - """Record for keeping track of open lockfiles (with reference counting).""" - - __slots__ = ("fh", "key", "refs") - - def __init__(self, fh: IO[bytes], key: DevIno): - self.fh = fh - self.key = key # (dev, ino) - self.refs = 0 - - -class OpenFileTracker: - """Track open lockfiles by inode, to minimize the number of open file descriptors. - - ``fcntl`` locks are associated with an inode. If a process closes *any* file descriptor for an - inode, all fcntl locks the process holds on that inode are released, even if other descriptors - for the same inode are still open. - - To avoid accidentally dropping locks we keep at most one open file descriptor per inode and - reference-count it. The descriptor is only closed when the reference count reaches zero (i.e. - no ``Lock`` in this process still needs it). - - Descriptors are *not* released on unlock; they are kept alive across lock/unlock cycles so that - the next lock operation can skip re-opening the file. ``PosixBackend._ensure_valid_handle`` - re-validates the on-disk inode before each lock operation and drops a stale descriptor when - the file was deleted and replaced. - """ - - def __init__(self): - self._descriptors: Dict[DevIno, OpenFile] = {} - - def get_ref_for_inode(self, key: DevIno) -> Optional[OpenFile]: - """Fast lookup: do we already have this inode open?""" - return self._descriptors.get(key) - - def create_and_track(self, path: str) -> OpenFile: - """Slow path: Open file, handle directory creation, track it.""" - # Open the file and create it if it doesn't exist (incl. directories). - try: - try: - fd = os.open(path, os.O_RDWR | os.O_CREAT) - mode = "rb+" - except PermissionError: - fd = os.open(path, os.O_RDONLY) - mode = "rb" - except OSError as e: - if e.errno != errno.ENOENT: - raise - # Directory missing, create and retry - try: - os.makedirs(os.path.dirname(path), exist_ok=True) - fd = os.open(path, os.O_RDWR | os.O_CREAT) - except OSError: - raise CantCreateLockError(path) - mode = "rb+" - - # Get file identifier (device, inode) for tracking. - stat = os.fstat(fd) - key = (stat.st_dev, stat.st_ino) - - # Did we open a file we already track, e.g. a symlink to existing tracker file. - if key in self._descriptors: - os.close(fd) - existing = self._descriptors[key] - existing.refs += 1 - return existing - - # Track the new file. - fh = os.fdopen(fd, mode) - obj = OpenFile(fh, key) - obj.refs += 1 - self._descriptors[key] = obj - return obj - - def release(self, open_file: OpenFile): - """Decrement the reference count and close the file handle when it reaches zero.""" - open_file.refs -= 1 - if open_file.refs <= 0: - if self._descriptors.get(open_file.key) is open_file: - del self._descriptors[open_file.key] - open_file.fh.close() - - def purge(self): - """Close all tracked file descriptors and clear the cache.""" - for open_file in self._descriptors.values(): - open_file.fh.close() - self._descriptors.clear() - - -#: Open file descriptors for locks in this process. Used to prevent one process -#: from opening the sam file many times for different byte range locks -FILE_TRACKER = OpenFileTracker() - - def _attempts_str(wait_time, nattempts): # Don't print anything if we succeeded on the first try if nattempts <= 1: @@ -156,195 +69,12 @@ def _attempts_str(wait_time, nattempts): return " after {} and {}".format(lang.pretty_seconds(wait_time), attempts) -class LockType: - READ = 0 - WRITE = 1 - - @staticmethod - def to_str(tid): - ret = "READ" - if tid == LockType.WRITE: - ret = "WRITE" - return ret - - @staticmethod - def to_module(tid): - lock = fcntl.LOCK_SH - if tid == LockType.WRITE: - lock = fcntl.LOCK_EX - return lock - - @staticmethod - def is_valid(op: int) -> bool: - return op == LockType.READ or op == LockType.WRITE - - -class PosixBackend: - """fcntl-based lock backend for POSIX systems.""" - - def __init__(self, path: str, start: int, length: int, debug: bool = False) -> None: - self.path = path - self._start = start - self._length = length - self.debug = debug - self._file_ref: Optional[OpenFile] = None - self._cached_key: Optional[DevIno] = None - # PID and host of the lock holder (only used in debug mode) - self.pid: Optional[int] = None - self.old_pid: Optional[int] = None - self.host: Optional[str] = None - self.old_host: Optional[str] = None - - def __getstate__(self): - state = self.__dict__.copy() - del state["_file_ref"] - del state["_cached_key"] - return state - - def __setstate__(self, state): - self.__dict__.update(state) - self._file_ref = None - self._cached_key = None - - def _ensure_valid_handle(self) -> IO[bytes]: - """Return a valid file handle for the lock file, opening or re-opening as needed. - - On the happy path this costs a single ``os.stat`` syscall: if the inode on disk matches - ``_cached_key``, the already-open file handle is returned immediately. - - If the inode changed (the lock file was deleted and replaced by another process), the stale - reference is released and a fresh one is obtained. If the file does not exist yet it is - created (along with any missing parent directories). - """ - try: - # Check what is currently on disk. This is the only syscall in the happy path. - stat_res = os.stat(self.path) - current_key = (stat_res.st_dev, stat_res.st_ino) - - # Double-check that our cache corresponds the file on disk. - if self._file_ref and not self._file_ref.fh.closed: - if self._cached_key == current_key: - return self._file_ref.fh - - # Stale path: file was deleted and replaced on disk. - FILE_TRACKER.release(self._file_ref) - self._file_ref = None - - # Get reference to the verified inode from the tracker if it exist, or a new one. - existing_ref = FILE_TRACKER.get_ref_for_inode(current_key) - if existing_ref: - self._file_ref = existing_ref - self._file_ref.refs += 1 - else: - # We don't have it tracked, so we need to open and track it ourselves. - self._file_ref = FILE_TRACKER.create_and_track(self.path) - except OSError as e: - # Re-raise all errors except for "file not found". - if e.errno != errno.ENOENT: - raise - - # File was not found, so remove it from our cache. - if self._file_ref: - FILE_TRACKER.release(self._file_ref) - self._file_ref = None - - self._file_ref = FILE_TRACKER.create_and_track(self.path) - - # Update our local cache of what we hold - self._cached_key = self._file_ref.key - - return self._file_ref.fh - - def prepare(self, op: int) -> None: - """Ensure the lock file is open; raise if a write lock is requested on a read-only file.""" - fh = self._ensure_valid_handle() - if LockType.to_module(op) == fcntl.LOCK_EX and fh.mode == "rb": - raise LockROFileError(self.path) - - def poll(self, op: int) -> bool: - """Attempt to acquire the lock in a non-blocking manner. Return whether - the locking attempt succeeds - """ - assert self._file_ref is not None, "cannot poll a lock without the file being set" - fh = self._file_ref.fh.fileno() - module_op = LockType.to_module(op) - try: - # Try to get the lock (will raise if not available.) - fcntl.lockf(fh, module_op | fcntl.LOCK_NB, self._length, self._start, os.SEEK_SET) - - # help for debugging distributed locking - if self.debug: - # All locks read the owner PID and host - self._read_log_debug_data() - tty.debug( - "{0} locked {1} [{2}:{3}] (owner={4})".format( - LockType.to_str(op), self.path, self._start, self._length, self.pid - ), - level=2, - ) - - # Exclusive locks write their PID/host - if module_op == fcntl.LOCK_EX: - self._write_log_debug_data() - - return True - - except OSError as e: - # EAGAIN and EACCES == locked by another process (so try again) - if e.errno not in (errno.EAGAIN, errno.EACCES): - raise - - return False - - def _read_log_debug_data(self) -> None: - """Read PID and host data out of the file if it is there.""" - assert self._file_ref is not None, "cannot read debug log without the file being set" - self.old_pid = self.pid - self.old_host = self.host - - self._file_ref.fh.seek(0) - line = self._file_ref.fh.read() - if line: - pid, host = line.decode("utf-8").strip().split(",") - _, _, pid = pid.rpartition("=") - _, _, self.host = host.rpartition("=") - self.pid = int(pid) - - def _write_log_debug_data(self) -> None: - """Write PID and host data to the file, recording old values.""" - assert self._file_ref is not None, "cannot write debug log without the file being set" - self.old_pid = self.pid - self.old_host = self.host - - self.pid = os.getpid() - self.host = socket.gethostname() - - # write pid, host to disk to sync over FS - self._file_ref.fh.seek(0) - self._file_ref.fh.write(f"pid={self.pid},host={self.host}".encode("utf-8")) - self._file_ref.fh.truncate() - self._file_ref.fh.flush() - os.fsync(self._file_ref.fh.fileno()) - - def release(self) -> None: - """Releases a lock using POSIX locks (``fcntl.lockf``) - - Releases the lock regardless of mode. Note that read locks may be masquerading as write - locks, but this removes either. - """ - assert self._file_ref is not None, "cannot unlock without the file being set" - fcntl.lockf( - self._file_ref.fh.fileno(), fcntl.LOCK_UN, self._length, self._start, os.SEEK_SET - ) - - def cleanup(self, path: str) -> None: - """Remove the lock file.""" - os.unlink(path) - - -class DummyBackend: +class DummyBackend(GenericLockBackend): """No-op lock backend: all operations succeed without acquiring any real locks.""" + def __init__(self) -> None: # doesn't need path/start/length: nothing is ever tracked + pass + def prepare(self, op: int) -> None: pass @@ -358,6 +88,22 @@ def cleanup(self, path: str) -> None: pass +#: Every backend derives from GenericLockBackend, and ``Lock`` only ever calls the methods +#: defined there -- using the base class here (rather than a Union of concrete backends) avoids +#: needing to reference the Windows-only ``WindowsBackend`` outside of a literal +#: ``sys.platform``-guarded import, which is what lets type checkers running on non-Windows +#: platforms (e.g. Read the Docs) skip ``spack.util.lock_windows`` entirely. +BackendType = GenericLockBackend + + +def platform_lock_backend(path, start, length, debug) -> BackendType: + """Per platform dispatch for lock backend implementation""" + if sys.platform == "win32": + return WindowsBackend(path, start, length, debug=debug) + else: + return PosixBackend(path, start, length, debug=debug) + + class Lock: """This is an implementation of a filesystem lock using Python's lockf. @@ -402,15 +148,17 @@ def __init__( desc: optional debug message lock description, which is helpful for distinguishing between different Spack locks. enable: when False, swap in a no-op backend so all lock operations succeed - without acquiring a real filesystem lock. Always disabled on Windows. + without acquiring a real filesystem lock. """ self.path = path self._reads = 0 self._writes = 0 - # byte range parameters + # byte range parameters. A zero length means "lock to the end of the file" on POSIX, + # but Windows has no such convention -- LockFileEx always needs an explicit range -- so + # a length of 0 is normalized to WHOLE_FILE_RANGE there. self._start = start - self._length = length + self._length = length or WHOLE_FILE_RANGE # enable debug mode self.debug = debug @@ -423,9 +171,9 @@ def __init__( # user sets a timeout for each attempt) self.default_timeout = default_timeout or None - if sys.platform != "win32" and enable: - self.backend: Union[PosixBackend, DummyBackend] = PosixBackend( - path, start, length, debug=debug + if enable: + self.backend: BackendType = platform_lock_backend( + path, start, self._length, debug=debug ) else: self.backend = DummyBackend() @@ -481,6 +229,10 @@ def __setstate__(self, state): self._reads = 0 self._writes = 0 + def _poll_lock(self, op: int) -> bool: + """Direct pass-through to the backend's single non-blocking lock attempt.""" + return self.backend.poll(op) + def _lock(self, op: int, timeout: Optional[float] = None) -> Tuple[float, int]: """This takes a lock using POSIX locks (``fcntl.lockf``). @@ -575,7 +327,12 @@ def _reaffirm_lock(self) -> None: """Fork-safety: always re-affirm the lock with one non-blocking attempt. In the same process, re-locking an already-held byte range succeeds instantly (POSIX). In a forked child that doesn't own the POSIX lock, the call fails immediately and we raise. Use WRITE - if we hold an exclusive lock so we don't accidentally downgrade it.""" + if we hold an exclusive lock so we don't accidentally downgrade it. + + No-op on Windows (Spawn only) + """ + if sys.platform == "win32": + return if self._writes > 0: op = LockType.WRITE elif self._reads > 0: @@ -607,12 +364,16 @@ def try_acquire_write(self) -> bool: """Non-blocking attempt to acquire an exclusive write lock. Returns True if the lock was acquired, False if it would block. + Handles three cases: no lock held (fresh acquire), read held (upgrade -- fully replaces + the read with a write, unlike the nested behavior of ``acquire_write``), or write already + held (nested). """ if self._writes == 0: self.backend.prepare(LockType.WRITE) if not self.backend.poll(LockType.WRITE): return False - self._writes += 1 + self._reads = 0 + self._writes = 1 self._log_acquired("WRITE LOCK", 0, 1) return True else: @@ -659,7 +420,7 @@ def upgrade_read_to_write(self, timeout: Optional[float] = None) -> None: """ timeout = timeout or self.default_timeout - if self._reads == 1 and self._writes == 0: + if self._reads >= 1 and self._writes == 0: self._log_upgrading() # can raise LockError. wait_time, nattempts = self._lock(LockType.WRITE, timeout=timeout) @@ -821,8 +582,15 @@ def __init__( self._release_fn = release def __enter__(self): - if self._enter() and self._acquire_fn: - return self._acquire_fn() + entered = self._enter() + if entered and self._acquire_fn: + try: + return self._acquire_fn() + except BaseException: + # If __enter__ raises, Python never calls __exit__, so the lock _enter() just + # acquired would otherwise leak. + self._exit(None) + raise def __exit__( self, @@ -977,10 +745,6 @@ def __exit__( return super().__exit__(exc_type, exc_value, traceback) -class LockError(Exception): - """Raised for any errors related to locks.""" - - class LockDowngradeError(LockError): """Raised when unable to downgrade from a write to a read lock.""" @@ -1012,24 +776,3 @@ class LockUpgradeError(LockError): def __init__(self, path: str) -> None: msg = "Cannot upgrade lock from read to write on file: %s" % path super().__init__(msg) - - -class LockPermissionError(LockError): - """Raised when there are permission issues with a lock.""" - - -class LockROFileError(LockPermissionError): - """Tried to take an exclusive lock on a read-only file.""" - - def __init__(self, path: str) -> None: - msg = "Can't take write lock on read-only file: %s" % path - super().__init__(msg) - - -class CantCreateLockError(LockPermissionError): - """Attempt to create a lock in an unwritable location.""" - - def __init__(self, path: str) -> None: - msg = "cannot create lock '%s': " % path - msg += "file does not exist and location is not writable" - super().__init__(msg) diff --git a/lib/spack/spack/util/lock_common.py b/lib/spack/spack/util/lock_common.py new file mode 100644 index 00000000000000..a2fd6a225bbe25 --- /dev/null +++ b/lib/spack/spack/util/lock_common.py @@ -0,0 +1,285 @@ +# Copyright Spack Project Developers. See COPYRIGHT file for details. +# +# SPDX-License-Identifier: (Apache-2.0 OR MIT) + +"""Platform-neutral plumbing shared by ``spack.util.lock``'s backends: the open-file tracker, +the ``LockType`` base, ``GenericLockBackend``, and the handful of lock exceptions backend code +itself needs to raise. +""" + +import errno +import os +import socket +from typing import IO, Dict, Optional, Tuple + +DevIno = Tuple[int, int] # (st_dev, st_ino) from os.stat_result + + +class OpenFile: + """Record for keeping track of open lockfiles (with reference counting).""" + + __slots__ = ("fh", "key", "refs") + + def __init__(self, fh: IO[bytes], key: DevIno): + self.fh = fh + self.key = key # (dev, ino) + self.refs = 0 + + +class OpenFileTracker: + """Track open lockfiles by inode, to minimize the number of open file descriptors. + + ``fcntl`` locks are associated with an inode. If a process closes *any* file descriptor for an + inode, all fcntl locks the process holds on that inode are released, even if other descriptors + for the same inode are still open. + + To avoid accidentally dropping locks we keep at most one open file descriptor per inode and + reference-count it. The descriptor is only closed when the reference count reaches zero (i.e. + no ``Lock`` in this process still needs it). + + Descriptors are *not* released on unlock; they are kept alive across lock/unlock cycles so that + the next lock operation can skip re-opening the file. ``PosixBackend._ensure_valid_handle`` + re-validates the on-disk inode before each lock operation and drops a stale descriptor when + the file was deleted and replaced. + """ + + def __init__(self): + self._descriptors: Dict[DevIno, OpenFile] = {} + + def get_ref_for_inode(self, key: DevIno) -> Optional[OpenFile]: + """Fast lookup: do we already have this inode open?""" + return self._descriptors.get(key) + + def create_and_track(self, path: str) -> OpenFile: + """Slow path: Open file, handle directory creation, track it.""" + # Open the file and create it if it doesn't exist (incl. directories). + try: + try: + fd = os.open(path, os.O_RDWR | os.O_CREAT) + mode = "rb+" + except PermissionError: + fd = os.open(path, os.O_RDONLY) + mode = "rb" + except OSError as e: + if e.errno != errno.ENOENT: + raise + # Directory missing, create and retry + try: + os.makedirs(os.path.dirname(path), exist_ok=True) + fd = os.open(path, os.O_RDWR | os.O_CREAT) + except OSError: + raise CantCreateLockError(path) + mode = "rb+" + + # Get file identifier (device, inode) for tracking. + stat = os.fstat(fd) + key = (stat.st_dev, stat.st_ino) + + # Did we open a file we already track, e.g. a symlink to existing tracker file. + if key in self._descriptors: + os.close(fd) + existing = self._descriptors[key] + existing.refs += 1 + return existing + + # Track the new file. + fh = os.fdopen(fd, mode) + obj = OpenFile(fh, key) + obj.refs += 1 + self._descriptors[key] = obj + return obj + + def release(self, open_file: OpenFile): + """Decrement the reference count and close the file handle when it reaches zero.""" + open_file.refs -= 1 + if open_file.refs <= 0: + if self._descriptors.get(open_file.key) is open_file: + del self._descriptors[open_file.key] + open_file.fh.close() + + def purge(self): + """Close all tracked file descriptors and clear the cache.""" + for open_file in self._descriptors.values(): + open_file.fh.close() + self._descriptors.clear() + + +#: Open file descriptors for locks in this process. Used to prevent one process +#: from opening the sam file many times for different byte range locks +FILE_TRACKER = OpenFileTracker() + + +class LockType: + """Platform-neutral lock operation identifiers. + """ + + READ = 0 + WRITE = 1 + + @staticmethod + def to_str(tid): + ret = "READ" + if tid == LockType.WRITE: + ret = "WRITE" + return ret + + @staticmethod + def is_valid(op: int) -> bool: + return op == LockType.READ or op == LockType.WRITE + + +class GenericLockBackend: + """Base class for platform lock backends. + + Handles bookkeeping shared by all backends: tracking the open file handle through + ``FILE_TRACKER`` and reading/writing the debug PID/host header. Subclasses implement the + actual OS-level locking primitives (``poll()`` and ``release()``). + """ + + def __init__(self, path: str, start: int, length: int, debug: bool = False) -> None: + self.path = path + self._start = start + self._length = length + self.debug = debug + self._file_ref: Optional[OpenFile] = None + self._cached_key: Optional[DevIno] = None + # PID and host of the lock holder (only used in debug mode) + self.pid: Optional[int] = None + self.old_pid: Optional[int] = None + self.host: Optional[str] = None + self.old_host: Optional[str] = None + + def __getstate__(self): + state = self.__dict__.copy() + del state["_file_ref"] + del state["_cached_key"] + return state + + def __setstate__(self, state): + self.__dict__.update(state) + self._file_ref = None + self._cached_key = None + + def _ensure_valid_handle(self) -> IO[bytes]: + """Return a valid file handle for the lock file, opening or re-opening as needed. + + On the happy path this costs a single ``os.stat`` syscall: if the inode on disk matches + ``_cached_key``, the already-open file handle is returned immediately. + + If the inode changed (the lock file was deleted and replaced by another process), the stale + reference is released and a fresh one is obtained. If the file does not exist yet it is + created (along with any missing parent directories). + """ + try: + # Check what is currently on disk. This is the only syscall in the happy path. + stat_res = os.stat(self.path) + current_key = (stat_res.st_dev, stat_res.st_ino) + + # Double-check that our cache corresponds the file on disk. + if self._file_ref and not self._file_ref.fh.closed: + if self._cached_key == current_key: + return self._file_ref.fh + + # Stale path: file was deleted and replaced on disk. + FILE_TRACKER.release(self._file_ref) + self._file_ref = None + + # Get reference to the verified inode from the tracker if it exist, or a new one. + existing_ref = FILE_TRACKER.get_ref_for_inode(current_key) + if existing_ref: + self._file_ref = existing_ref + self._file_ref.refs += 1 + else: + # We don't have it tracked, so we need to open and track it ourselves. + self._file_ref = FILE_TRACKER.create_and_track(self.path) + except OSError as e: + # Re-raise all errors except for "file not found". + if e.errno != errno.ENOENT: + raise + + # File was not found, so remove it from our cache. + if self._file_ref: + FILE_TRACKER.release(self._file_ref) + self._file_ref = None + + self._file_ref = FILE_TRACKER.create_and_track(self.path) + + # Update our local cache of what we hold + self._cached_key = self._file_ref.key + + return self._file_ref.fh + + def prepare(self, op: int) -> None: + """Ensure the lock file is open; raise if a write lock is requested on a read-only file.""" + fh = self._ensure_valid_handle() + + if op == LockType.WRITE and fh.mode == "rb": + # Attempt to upgrade to write lock w/a read-only file. + # If the file were writable, we'd have opened it rb+ + raise LockROFileError(self.path) + + def cleanup(self, path: str) -> None: + """Remove the lock file.""" + os.unlink(path) + + def _read_log_debug_data(self) -> None: + """Read PID and host data out of the file if it is there.""" + assert self._file_ref is not None, "cannot read debug log without the file being set" + + self.old_pid = self.pid + self.old_host = self.host + + self._file_ref.fh.seek(0) + line = self._file_ref.fh.read() + if line: + pid, host = line.decode("utf-8").strip().split(",") + _, _, pid = pid.rpartition("=") + _, _, self.host = host.rpartition("=") + self.pid = int(pid) + + def _write_log_debug_data(self) -> None: + """Write PID and host data to the file, recording old values.""" + assert self._file_ref is not None, "cannot write debug log without the file being set" + + self.old_pid = self.pid + self.old_host = self.host + + self.pid = os.getpid() + self.host = socket.gethostname() + # write pid, host to disk to sync over FS + self._file_ref.fh.seek(0) + self._file_ref.fh.write(f"pid={self.pid},host={self.host}".encode("utf-8")) + self._file_ref.fh.truncate() + self._file_ref.fh.flush() + os.fsync(self._file_ref.fh.fileno()) + + def poll(self, op: int) -> bool: + raise NotImplementedError + + def release(self) -> None: + raise NotImplementedError + + +class LockError(Exception): + """Raised for any errors related to locks.""" + + +class LockPermissionError(LockError): + """Raised when there are permission issues with a lock.""" + + +class LockROFileError(LockPermissionError): + """Tried to take an exclusive lock on a read-only file.""" + + def __init__(self, path: str) -> None: + msg = "Can't take write lock on read-only file: %s" % path + super().__init__(msg) + + +class CantCreateLockError(LockPermissionError): + """Attempt to create a lock in an unwritable location.""" + + def __init__(self, path: str) -> None: + msg = "cannot create lock '%s': " % path + msg += "file does not exist and location is not writable" + super().__init__(msg) diff --git a/lib/spack/spack/util/lock_posix.py b/lib/spack/spack/util/lock_posix.py new file mode 100644 index 00000000000000..3a7d4235de9591 --- /dev/null +++ b/lib/spack/spack/util/lock_posix.py @@ -0,0 +1,91 @@ +# Copyright Spack Project Developers. See COPYRIGHT file for details. +# +# SPDX-License-Identifier: (Apache-2.0 OR MIT) + +"""fcntl-based lock backend for POSIX systems, and the concrete ``LockType`` that supplies its +native flag values. +""" + +import sys + +if sys.platform == "win32": + # Also lets mypy skip this module when run on Windows. + raise ImportError("spack.util.lock_posix can only be imported on POSIX platforms") + +import errno +import fcntl +import os + +from spack.util import tty + +from .lock_common import GenericLockBackend +from .lock_common import LockType as _LockType + + +class LockType(_LockType): + LOCK_SH = fcntl.LOCK_SH + LOCK_EX = fcntl.LOCK_EX + LOCK_NB = fcntl.LOCK_NB + LOCK_UN = fcntl.LOCK_UN + LOCK_CATCH = OSError + + @staticmethod + def to_module(tid): + lock = LockType.LOCK_SH + if tid == LockType.WRITE: + lock = LockType.LOCK_EX + return lock + + +class PosixBackend(GenericLockBackend): + """fcntl-based lock backend for POSIX systems.""" + + def poll(self, op: int) -> bool: + """Attempt to acquire the lock in a non-blocking manner. Return whether + the locking attempt succeeds + """ + assert self._file_ref is not None, "cannot poll a lock without the file being set" + fh = self._file_ref.fh.fileno() + module_op = LockType.to_module(op) + + try: + # Try to get the lock (will raise if not available.) + fcntl.lockf(fh, module_op | LockType.LOCK_NB, self._length, self._start, os.SEEK_SET) + + # help for debugging distributed locking + if self.debug: + # All locks read the owner PID and host + self._read_log_debug_data() + tty.debug( + "{0} locked {1} [{2}:{3}] (owner={4})".format( + LockType.to_str(op), self.path, self._start, self._length, self.pid + ), + level=2, + ) + + # Exclusive locks write their PID/host + if op == LockType.WRITE: + self._write_log_debug_data() + + return True + + except LockType.LOCK_CATCH as e: + # EAGAIN and EACCES == locked by another process (so try again) + if self._lock_fail_condition(e): + raise + + return False + + def _lock_fail_condition(self, e) -> bool: + return e.errno not in (errno.EAGAIN, errno.EACCES) + + def release(self) -> None: + """Releases a lock using POSIX locks (``fcntl.lockf``) + + Releases the lock regardless of mode. Note that read locks may be masquerading as write + locks, but this removes either. + """ + assert self._file_ref is not None, "cannot unlock without the file being set" + fcntl.lockf( + self._file_ref.fh.fileno(), LockType.LOCK_UN, self._length, self._start, os.SEEK_SET + ) diff --git a/lib/spack/spack/util/lock_windows.py b/lib/spack/spack/util/lock_windows.py new file mode 100644 index 00000000000000..71fcc77a931742 --- /dev/null +++ b/lib/spack/spack/util/lock_windows.py @@ -0,0 +1,438 @@ +# Copyright Spack Project Developers. See COPYRIGHT file for details. +# +# SPDX-License-Identifier: (Apache-2.0 OR MIT) + +"""LockFileEx-based lock backend for Windows, and the ctypes kernel32 bindings it needs. +""" + +import sys + +if sys.platform != "win32": + # Also lets mypy skip this module when run on other platforms. + raise ImportError("spack.util.lock_windows can only be imported on Windows") + +import ctypes +import msvcrt +import os +from ctypes import wintypes +from typing import Dict, List, Optional + +from spack.util import tty + +from .lock_common import FILE_TRACKER, DevIno, GenericLockBackend +from .lock_common import LockType as _LockType + + +class LockType(_LockType): + # From the Windows SDK (winbase.h): not exposed by ctypes, so hardcoded here. + LOCK_SH = 0 # shared lock is the default (absence of the exclusive flag) + LOCK_EX = 0x00000002 # LOCKFILE_EXCLUSIVE_LOCK + LOCK_NB = 0x00000001 # LOCKFILE_FAIL_IMMEDIATELY + + @staticmethod + def to_module(tid): + lock = LockType.LOCK_SH + if tid == LockType.WRITE: + lock = LockType.LOCK_EX + return lock + + +class _OVERLAPPED(ctypes.Structure): + _fields_ = [ + ("Internal", ctypes.c_void_p), + ("InternalHigh", ctypes.c_void_p), + ("Offset", wintypes.DWORD), + ("OffsetHigh", wintypes.DWORD), + ("hEvent", wintypes.HANDLE), + ] + + +_kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + +_kernel32.LockFileEx.argtypes = [ + wintypes.HANDLE, + wintypes.DWORD, + wintypes.DWORD, + wintypes.DWORD, + wintypes.DWORD, + ctypes.POINTER(_OVERLAPPED), +] +_kernel32.LockFileEx.restype = wintypes.BOOL + +_kernel32.UnlockFileEx.argtypes = [ + wintypes.HANDLE, + wintypes.DWORD, + wintypes.DWORD, + wintypes.DWORD, + ctypes.POINTER(_OVERLAPPED), +] +_kernel32.UnlockFileEx.restype = wintypes.BOOL + +# winerror 32: "The process cannot access the file because it is being used by another +# process." +# winerror 33: "The process cannot access the file because another process has locked a +# portion of the file." +_ERROR_SHARING_VIOLATION = 32 +_ERROR_LOCK_VIOLATION = 33 + + +def _low_high(value): + low = value & 0xFFFFFFFF + high = (value >> 32) & 0xFFFFFFFF + return low, high + + +def _setup_overlapped(offset): + overlapped = _OVERLAPPED() + # hEvent needs to be null per LockFileEx/UnlockFileEx docs + overlapped.hEvent = 0 + overlapped.Offset, overlapped.OffsetHigh = _low_high(offset) + return overlapped + + +def _win_handle(fd: int) -> int: + """The raw Win32 HANDLE backing a Python file descriptor.""" + return msvcrt.get_osfhandle(fd) + + +def _win_lock_file_ex(handle: int, flags: int, low: int, high: int, overlapped) -> bool: + """Wraps ``LockFileEx``. Returns whether the lock was acquired: True on success, False if + the range is already locked by someone else (only possible when ``flags`` includes + ``LockType.LOCK_NB``). Raises ``OSError`` for any other failure. + """ + if _kernel32.LockFileEx(handle, flags, 0, low, high, ctypes.byref(overlapped)): + return True + err = ctypes.get_last_error() + if err in (_ERROR_SHARING_VIOLATION, _ERROR_LOCK_VIOLATION): + return False + raise ctypes.WinError(err) + + +def _win_unlock_file_ex(handle: int, low: int, high: int, overlapped) -> None: + """Wraps ``UnlockFileEx``. Raises ``OSError`` on failure.""" + if not _kernel32.UnlockFileEx(handle, 0, low, high, ctypes.byref(overlapped)): + raise ctypes.WinError(ctypes.get_last_error()) + + +class WindowsRangeLock: + """The one real ``LockFileEx`` lock for a given byte range in this process, shared by every + ``WindowsBackend`` that requests an overlapping range while it's held. See + ``WindowsRangeLockTracker`` for why this is needed. + """ + + __slots__ = ("start", "length", "mode", "refs") + + def __init__(self, start: int, length: int, mode: int): + self.start = start + self.length = length + #: LockType.READ or LockType.WRITE: the real mode currently held, via whichever handle + #: any attached backend happens to have open + self.mode = mode + self.refs = 1 + + +class WindowsRangeLockTracker: + """Tracks byte ranges the current process holds real Windows locks on, so that a second + ``Lock`` in the same process requesting an exactly overlapping range shares the *same* + real lock state instead of contending with its own process. + + Windows ``LockFileEx`` locks are scoped to the specific *handle* that acquired them: two + different handles regardless of process will compete for lock acquisition. + Spack's locking code (e.g. ``FailureTracker``/``SpecLocker`` re-checking a lock it + itself holds via a second, independent ``Lock`` object, or a plain upgrade/downgrade) + will cause same-process lock contention for the same lock, even with the same handle. + + Every ``WindowsBackend`` for a given range shares the exact same open file handle for the + underlying file, and this tracker records the one real lock ``mode`` currently held on + a range in addition to a ref count, so we can accurately track lock usages and update/change + the lock state for this handle/lock range as needed without contention. + """ + + def __init__(self): + self._groups: Dict[DevIno, List[WindowsRangeLock]] = {} + + @staticmethod + def _contains(outer_start: int, outer_length: int, start: int, length: int) -> bool: + return outer_start <= start and start + length <= outer_start + outer_length + + def find(self, key: DevIno, start: int, length: int) -> Optional[WindowsRangeLock]: + """Return the group already covering [start, start+length) in this process, if any.""" + for group in self._groups.get(key, []): + if self._contains(group.start, group.length, start, length): + return group + return None + + def register(self, key: DevIno, group: WindowsRangeLock) -> None: + """Record a freshly, really-acquired OS lock as a new group of one.""" + self._groups.setdefault(key, []).append(group) + + def forget(self, key: DevIno, group: WindowsRangeLock) -> None: + """Drop a group that no backend references anymore.""" + groups = self._groups.get(key, []) + if group in groups: + groups.remove(group) + if not groups: + self._groups.pop(key, None) + + +#: Tracks real Windows byte-range locks held by this process, to make same-process, +#: cross-handle lock requests behave like POSIX fcntl. +WINDOWS_RANGE_LOCK_TRACKER = WindowsRangeLockTracker() + + +class WindowsBackend(GenericLockBackend): + """LockFileEx-based lock backend for Windows. + + Like ``PosixBackend``, every ``WindowsBackend`` for a given file shares one open handle via + the process-wide ``FILE_TRACKER`` (see ``GenericLockBackend._ensure_valid_handle``) + That alone would be unsafe for Windows locking: unlike ``fcntl``, + ``LockFileEx``/``UnlockFileEx`` calls apply to whichever handle makes them, so two unrelated + ``Lock`` objects sharing a handle will compete for the "same" lock, and potentially deadlock. + ``WINDOWS_RANGE_LOCK_TRACKER`` tracks locks themselves, not FH like FILE_TRACKER, including the + handle, locking style, and ref counts, so multiple lock attempts from the same process on the + same range behavemore closely to posix locks. + + There are two other contexts in which Windows diverges from posix + + * *Atomic mode transitions.* ``fcntl`` can convert an already-held lock to a different mode + (shared <-> exclusive) in place, in a single call. ``LockFileEx`` cannot: there is no + "convert" operation. ``poll()`` detects a mode transition on an already-held range and + handles it internally: downgrading a held exclusive lock uses a stack-then-unlock-once + trick (see ``_downgrade_to_read``); upgrading a held shared lock uses a dedicated per-range + "gate" lock file to serialize against every other write attempt while the range is briefly, + fully released and retaken (see ``_upgrade_to_write``). + + * *Per-process (not per-handle) lock scoping.* ``fcntl`` locks are scoped to (process, + inode): a process can always freely take another lock, in any mode, on a range it already + holds, via any file descriptor. Because every same-process backend for a range shares the + same real handle and the same ``WindowsRangeLock``, a mode change made by any one of them + is immediately visible to all of them, matching that semantics exactly, with no separate + "anchor" handle to reason about. + + ``release()`` closes this backend's handle reference (via ``FILE_TRACKER``, closing the + underlying handle only once every backend sharing it has released) -- needed so a later + ``cleanup()``/``os.unlink()`` doesn't fail with WinError 32, "used by another process". + """ + + def __init__(self, path: str, start: int, length: int, debug: bool = False) -> None: + super().__init__(path, start, length, debug=debug) + #: the WindowsRangeLock this backend belongs to while it holds a lock, None otherwise. + self._lock_group: Optional[WindowsRangeLock] = None + #: lazily-created backend for this range's gate file, used only for upgrades. + self._gate: Optional["WindowsBackend"] = None + + def __getstate__(self): + state = super().__getstate__() + del state["_lock_group"] + del state["_gate"] + return state + + def __setstate__(self, state): + super().__setstate__(state) + self._lock_group = None + self._gate = None + + def __del__(self) -> None: + # Guard against a Lock being dropped without an explicit release + # release properly (respecting shared-group bookkeeping) so a + # later cleanup()/unlink() doesn't fail with WinError 32, and so we don't leave a + # same-process WindowsRangeLock group permanently over-referenced. + if self._file_ref is None: + return + try: + self.release() + except Exception: + try: + FILE_TRACKER.release(self._file_ref) + except Exception: + pass + self._file_ref = None + self._cached_key = None + + def poll(self, op: int) -> bool: + """Attempt to acquire the lock in ``op`` mode in a non-blocking manner. Return whether + the attempt succeeds. + """ + assert self._file_ref is not None, "cannot poll a lock without the file being set" + + if self._lock_group is not None: + if op == LockType.READ and self._lock_group.mode == LockType.WRITE: + self._downgrade_to_read() + return True + if op == self._lock_group.mode: + return True # already holding at least as much as requested + return self._upgrade_to_write() # op == WRITE, mode == READ + + key = self._file_ref.key + group = WINDOWS_RANGE_LOCK_TRACKER.find(key, self._start, self._length) + if group is not None: + self._lock_group = group + if op == LockType.READ or group.mode == LockType.WRITE: + group.refs += 1 + return True + if self._upgrade_to_write(): + group.refs += 1 + return True + self._lock_group = None + return False + + return self._acquire_new(key, op) + + def _acquire_new(self, key: DevIno, op: int) -> bool: + """Non-blocking attempt to take a fresh real OS f-lock on this range: no backend in this + process currently holds anything overlapping it. + """ + assert self._file_ref is not None + + handle = _win_handle(self._file_ref.fh.fileno()) + module_op = LockType.to_module(op) + overlapped = _setup_overlapped(self._start) + range_low, range_high = _low_high(self._length) + + if not _win_lock_file_ex( + handle, module_op | LockType.LOCK_NB, range_low, range_high, overlapped + ): + return False + + # help for debugging distributed locking + if self.debug: + # All locks read the owner PID and host + self._read_log_debug_data() + tty.debug( + "{0} locked {1} [{2}:{3}] (owner={4})".format( + LockType.to_str(op), self.path, self._start, self._length, self.pid + ), + level=2, + ) + + # Exclusive locks write their PID/host + if op == LockType.WRITE: + self._write_log_debug_data() + + self._lock_group = WindowsRangeLock(self._start, self._length, op) + WINDOWS_RANGE_LOCK_TRACKER.register(key, self._lock_group) + return True + + def _gate_backend(self) -> "WindowsBackend": + """The backend for this range's dedicated upgrade "gate" file (see + ``_upgrade_to_write``), created and opened lazily on first use. + """ + if self._gate is None: + self._gate = WindowsBackend( + self.path + ".gate_lock", self._start, self._length, debug=self.debug + ) + return self._gate + + def _upgrade_to_write(self) -> bool: + """Single non-blocking attempt to upgrade this range's currently-held real read lock to + a write lock. + + ``LockFileEx`` has no atomic convert, and naively dropping the read and retaking + exclusive would subject spack to races. + Instead, take a "gate" lock first. Every write acquisition on this + range takes the same gate first, so while we hold the gate, no other same-process writer + can be mid-attempt, and only after actually dropping the read do we retake exclusive. + If that fails, restore the read, which we can guaruntee as we are still behind the gate. + + Note: on Windows the effective timeout for a blocking caller is up to 2x their requested + timeout, because ``Lock._lock``'s retry loop calls this once per attempt, and each + attempt both polls the gate and (if granted) polls the primary range. + + Note: the ``.gate_lock`` sidecar file this creates is cleaned up the same way as the + primary lock file, see ``WindowsBackend.cleanup``. + """ + assert self._file_ref is not None + group = self._lock_group + assert group is not None and group.mode == LockType.READ + + gate = self._gate_backend() + gate.prepare(LockType.WRITE) + if not gate.poll(LockType.WRITE): + return False + + try: + handle = _win_handle(self._file_ref.fh.fileno()) + range_low, range_high = _low_high(self._length) + + _win_unlock_file_ex(handle, range_low, range_high, _setup_overlapped(self._start)) + if _win_lock_file_ex( + handle, + LockType.LOCK_EX | LockType.LOCK_NB, + range_low, + range_high, + _setup_overlapped(self._start), + ): + group.mode = LockType.WRITE + return True + + # A plain reader (no gate needed) raced in during the drop: restore our read. + _win_lock_file_ex( + handle, + LockType.LOCK_SH | LockType.LOCK_NB, + range_low, + range_high, + _setup_overlapped(self._start), + ) + return False + finally: + gate.release() + + def _downgrade_to_read(self) -> None: + """Convert this range's held real exclusive lock to a shared lock, without ever fully + releasing it (so no other process can grab exclusive access in the gap caused by a + lack of atomic lock transitions). + + The win32 locking API has no atomic "convert" operation. But overlapping locks are allowed + on the same range from the same handle if the exclusive lock is taken first, + and Windows removes locks in FIFO order. So, stack a shared lock on top of the exclusive + one already held (this blocking call returns immediately, the handle already owns the + range, so there's nothing to wait for), then remove one lock, which drops the older + exclusive lock and leaves the shared one in place. Always succeeds immediately: no gate + is needed here, unlike upgrading, because the range is never actually unlocked. + """ + assert self._file_ref is not None + group = self._lock_group + assert group is not None and group.mode == LockType.WRITE + + handle = _win_handle(self._file_ref.fh.fileno()) + range_low, range_high = _low_high(self._length) + _win_lock_file_ex( + handle, LockType.LOCK_SH, range_low, range_high, _setup_overlapped(self._start) + ) + _win_unlock_file_ex(handle, range_low, range_high, _setup_overlapped(self._start)) + group.mode = LockType.READ + + def release(self) -> None: + """Release this backend's interest in the lock, and its ``FILE_TRACKER`` handle + reference so a later ``cleanup()`` can unlink (closing the underlying handle only once + every backend sharing it has released it). + + If this was the last backend relying on this range's ``WindowsRangeLock``, performs the + real OS-level unlock. + """ + assert self._file_ref is not None, "cannot unlock without the file being set" + key = self._file_ref.key + group = self._lock_group + self._lock_group = None + + if group is not None: + group.refs -= 1 + if group.refs <= 0: + WINDOWS_RANGE_LOCK_TRACKER.forget(key, group) + handle = _win_handle(self._file_ref.fh.fileno()) + range_low, range_high = _low_high(self._length) + _win_unlock_file_ex(handle, range_low, range_high, _setup_overlapped(self._start)) + + FILE_TRACKER.release(self._file_ref) + self._file_ref = None + self._cached_key = None + + def cleanup(self, path: str) -> None: + """Remove the lock file, and its ``.gate_lock`` sidecar (see ``_upgrade_to_write``) if + one was ever created for it. + """ + super().cleanup(path) + try: + os.unlink(path + ".gate_lock") + except FileNotFoundError: + pass diff --git a/share/spack/qa/configuration/windows_locking_config.yaml b/share/spack/qa/configuration/windows_locking_config.yaml new file mode 100644 index 00000000000000..266c7c42df93b0 --- /dev/null +++ b/share/spack/qa/configuration/windows_locking_config.yaml @@ -0,0 +1,8 @@ +config: + locks: true + install_tree: + root: $spack\opt\spack + projections: + all: '${ARCHITECTURE}\${COMPILERNAME}-${COMPILERVER}\${PACKAGE}-${VERSION}-${HASH}' + build_stage: + - ~/.spack/stage