diff --git a/hgitaly/tests/test_workdir.py b/hgitaly/tests/test_workdir.py index fb44e1f5c60268bfe416988e703abd9fb913a959_aGdpdGFseS90ZXN0cy90ZXN0X3dvcmtkaXIucHk=..5976d391ba25087bf3b32e96542eb77e2446520f_aGdpdGFseS90ZXN0cy90ZXN0X3dvcmtkaXIucHk= 100644 --- a/hgitaly/tests/test_workdir.py +++ b/hgitaly/tests/test_workdir.py @@ -10,6 +10,7 @@ import warnings +from mercurial import error from hgext3rd.heptapod.branch import set_default_gitlab_branch import pytest @@ -30,6 +31,8 @@ workdirs_gc, ) +parametrize = pytest.mark.parametrize + @pytest.fixture def wd_fixture(tmpdir): @@ -313,9 +316,13 @@ assert pipe.recv() == expected +ROSTER_LOCK_TIMEOUT = 0.1 +ROSTER_LOCK_ATTEMPTS_DELAY = ROSTER_LOCK_TIMEOUT / 10 + + def locking_subpro(repo_path, conn): """Subprocess taking the roster lock and then taking commands from a pipe. The lock is taken at startup after sending the initial message and is held until the 'shutdown' message is received """ @@ -316,8 +323,13 @@ def locking_subpro(repo_path, conn): """Subprocess taking the roster lock and then taking commands from a pipe. The lock is taken at startup after sending the initial message and is held until the 'shutdown' message is received """ + cleanup_on_signal(signal.SIGTERM) + cleanup_on_signal(signal.SIGINT) + from hgitaly import procutil + procutil.IS_CHILD_PROCESS = True + repo = LocalRepoWrapper.load(repo_path).repo conn.send('started') @@ -322,24 +334,27 @@ repo = LocalRepoWrapper.load(repo_path).repo conn.send('started') - with locked_roster(repo) as (inf, outf): - conn.send('locked') - while True: - msg = conn.recv() - if msg == 'shutdown': - conn.send('bye') - conn.close() - return - if isinstance(msg, tuple) and msg[0] == 'write': - outf.write(msg[1].encode('utf8')) - # atomictempfile does not implement flush - conn.send('written') - if msg == 'read': - if inf is None: - conn.send(None) - else: - pos = inf.tell() - inf.seek(0) - conn.send(inf.read().decode('utf8')) - inf.seek(pos) + try: + with locked_roster(repo, timeout=ROSTER_LOCK_TIMEOUT) as (inf, outf): + conn.send('locked') + while True: + msg = conn.recv() + if msg == 'shutdown': + conn.send('bye') + conn.close() + return + if isinstance(msg, tuple) and msg[0] == 'write': + outf.write(msg[1].encode('utf8')) + # atomictempfile does not implement flush + conn.send('written') + if msg == 'read': + if inf is None: + conn.send(None) + else: + pos = inf.tell() + inf.seek(0) + conn.send(inf.read().decode('utf8')) + inf.seek(pos) + except error.LockHeld: + conn.send('timeout') @@ -344,9 +359,10 @@ -def test_locked_roster(wd_fixture): +@parametrize('lock_attempts', (1, 4)) +def test_locked_roster(wd_fixture, lock_attempts): wrapper = wd_fixture[1] repo_path = wrapper.path pipe1, child_pipe1 = Pipe() pipe2, child_pipe2 = Pipe() pipe3, child_pipe3 = Pipe() @@ -347,9 +363,10 @@ wrapper = wd_fixture[1] repo_path = wrapper.path pipe1, child_pipe1 = Pipe() pipe2, child_pipe2 = Pipe() pipe3, child_pipe3 = Pipe() + pipe4, child_pipe4 = Pipe() proc1 = Process(target=locking_subpro, args=(repo_path, child_pipe1)) proc2 = Process(target=locking_subpro, args=(repo_path, child_pipe2)) proc3 = Process(target=locking_subpro, args=(repo_path, child_pipe3)) @@ -353,7 +370,8 @@ proc1 = Process(target=locking_subpro, args=(repo_path, child_pipe1)) proc2 = Process(target=locking_subpro, args=(repo_path, child_pipe2)) proc3 = Process(target=locking_subpro, args=(repo_path, child_pipe3)) - procs = [proc1, proc2, proc3] + proc4 = Process(target=locking_subpro, args=(repo_path, child_pipe4)) + procs = [proc1, proc2, proc3, proc4] try: # proc1 starts, write a line, but does not see it in its input file @@ -369,7 +387,8 @@ # proc2 starts but cannot acquire the lock yet proc2.start() assert_recv(pipe2, 'started') - assert not pipe2.poll(timeout=0.1) + assert not pipe2.poll( + timeout=ROSTER_LOCK_ATTEMPTS_DELAY * lock_attempts) # shutting down proc1 pipe1.send('shutdown') @@ -389,6 +408,6 @@ pipe2.send('read') assert_recv(pipe2, 'content1') - # proc3 starts, cannot acquire the lock either + # proc3 starts, cannot acquire the lock immediately either proc3.start() assert_recv(pipe3, 'started') @@ -393,6 +412,6 @@ proc3.start() assert_recv(pipe3, 'started') - assert not pipe3.poll(timeout=0.1) + assert not pipe3.poll(timeout=ROSTER_LOCK_ATTEMPTS_DELAY) # after proc2 shutdown, proc3 sees the new content pipe2.send('shutdown') @@ -402,6 +421,11 @@ pipe3.send('read') assert_recv(pipe3, 'content2') + # proc4 starts but proc3 does not release the lock in time + proc4.start() + assert_recv(pipe4, 'started') + assert_recv(pipe4, 'timeout') + pipe3.send('shutdown') assert_recv(pipe3, 'bye') finally: @@ -412,6 +436,57 @@ proc.join() +def test_roster_lock_breaking(wd_fixture): + wrapper = wd_fixture[1] + repo_path = wrapper.path + + pipe1, child_pipe1 = Pipe() + pipe2, child_pipe2 = Pipe() + proc1 = Process(target=locking_subpro, args=(repo_path, child_pipe1)) + proc2 = Process(target=locking_subpro, args=(repo_path, child_pipe2)) + procs = [proc1, proc2] + + # let's grab the lock from the main tests process, which is not allowed + # to take it, as it is not one of the HGitaly worker processes. This + # simulates the case where the PID has been reused: there *is* a process + # with that pid. + try: + with locked_roster(wrapper.repo) as (inf, outf): + # proc1 takes the lock, ignoring the lock held with an invalid PID + proc1.start() + assert_recv(pipe1, 'started') + assert_recv(pipe1, 'locked') + pipe1.send(('write', 'content1')) + assert_recv(pipe1, 'written') + + # of course the lock taken by proc1 blocks proc2 + # Note that exiting normally the `locked_roster` context manager + # of the main process would release the lock, even if held by + # proc1, which looks bad, but is irrelevant: in actual operation, + # roster locks have to be broken if the holding process have died + # abruptly enough not to have been able to release the lock. + proc2.start() + assert_recv(pipe2, 'started') + assert not pipe2.poll(timeout=ROSTER_LOCK_ATTEMPTS_DELAY) + + # shutting down proc1 + pipe1.send('shutdown') + assert_recv(pipe1, 'bye') + proc1.join() + + # now that proc1 has released the lock, proc2 acquires it + # and sees the write done by proc1 + assert_recv(pipe2, 'locked') + pipe2.send('read') + assert_recv(pipe2, 'content1') + finally: + # avoid blocking the test run if there are errors + for proc in procs: + if proc.is_alive(): + proc.terminate() + proc.join() + + def reserving_subpro(wds_root, repo_path, conn): cleanup_on_signal(signal.SIGTERM) cleanup_on_signal(signal.SIGINT) diff --git a/hgitaly/workdir.py b/hgitaly/workdir.py index fb44e1f5c60268bfe416988e703abd9fb913a959_aGdpdGFseS93b3JrZGlyLnB5..5976d391ba25087bf3b32e96542eb77e2446520f_aGdpdGFseS93b3JrZGlyLnB5 100644 --- a/hgitaly/workdir.py +++ b/hgitaly/workdir.py @@ -6,6 +6,7 @@ # SPDX-License-Identifier: GPL-2.0-or-later from contextlib import contextmanager import attr +import errno import logging import os from pathlib import Path @@ -17,4 +18,5 @@ from mercurial import ( cmdutil, commands, + error, hg, @@ -20,4 +22,5 @@ hg, + lock, ) from mercurial.repoview import _filteredrepotypes @@ -132,4 +135,68 @@ release_workdir(repo, wd) +class rosterlock(lock.lock): + + def _lockshouldbebroken(self, locker): + pid = int(locker.split(b":", 1)[-1].decode('ascii').strip()) + return not is_current_service_process(pid) + + +def trylock(ui, vfs, lockname, timeout, warntimeout, *args, step=0.1, + **kwargs): + """return an acquired lock or raise an a LockHeld exception + + This function is responsible to issue logs about + the held lock while trying to acquires it. + + Derived from Mercurial's `lock.trylock`, with these differences: + + - using :data:`logger` + - using :class:`rosterlock` + - configurable sleep interval (param ``step``), both because roster + file operations are designed to be fast, and to reach the warntimeout + in tests without overlong sleeps. + - timeouts always exist, since server operation must not be stalled + forever. + - `acquirefn` is ignored, as we don't need it for the roster lock. + """ + + def log(level, locker): + """log the "waiting on lock" message through at the given level""" + pid = locker.split(b":", 1)[-1].decode('ascii') + logger.log(level, "waiting for lock on %r held by process %r", + lk.desc, pid) + + lk = rosterlock(vfs, lockname, 0, *args, dolock=False, **kwargs) + + debugidx = 0 + warningidx = int(warntimeout / step) + + delay = 0 + while True: + try: + lk._trylock() + break + except error.LockHeld as inst: + if delay == debugidx: + log(logging.DEBUG, inst.locker) + if delay == warningidx: + log(logging.WARNING, inst.locker) + if timeout <= delay * step: + raise error.LockHeld( + errno.ETIMEDOUT, inst.filename, lk.desc, inst.locker + ) + time.sleep(step) + delay += 1 + + lk.delay = delay + if delay: + if 0 <= warningidx <= delay: + logger.warning("got lock after %.2f seconds", delay * step) + else: + logger.debug("got lock after %.2f seconds", delay * step) + + return lk + + @contextmanager @@ -135,5 +202,5 @@ @contextmanager -def locked_roster(repo): +def locked_roster(repo, timeout=1): """Locked read/write access to the repo working directories roster file. This context manager provides a pair of open files. @@ -142,6 +209,10 @@ The lock is not reentrant, which is good enough for this simple need of a very short-lived lock protecting both readind and writing. + + :param timeout: maximum time to wait until the roster lock is acquired. + The default value is intended for production, tests will set it to + shorter values. """ vfs = repo.vfs # TODO Mercurial lock system does not allow to customize its breaking @@ -153,14 +224,21 @@ # it is dubious that HGitaly ever gets exclusive access to Mercurial # content (HTTP pushes could be handled if HGitaly eventually manages # the hgwebdir services, but SSH pushes would not). - with repo._lock( - vfs, - lockname=ROSTER_LOCK_NAME, - releasefn=None, - acquirefn=None, - wait=True, - desc=b'Working directories roster file for %s' % repo.root - ): + ui = repo.ui + warntimeout = 3 * timeout / 10 + # internal config: ui.signal-safe-lock + signalsafe = ui.configbool(b'ui', b'signal-safe-lock') + + with trylock(ui, vfs, + lockname=ROSTER_LOCK_NAME, + timeout=timeout, + warntimeout=warntimeout, + releasefn=None, + acquirefn=None, + desc=b'Working directories roster file for %s' % repo.root, + signalsafe=signalsafe, + step=timeout / 10, + ): try: inf = vfs(ROSTER_FILE_NAME, b'rb') except FileNotFoundError: