# HG changeset patch # User Georges Racinet <georges.racinet@octobus.net> # Date 1682896635 -7200 # Mon May 01 01:17:15 2023 +0200 # Node ID 5976d391ba25087bf3b32e96542eb77e2446520f # Parent fb44e1f5c60268bfe416988e703abd9fb913a959 workdir: customized roster lock for breaking logic as in reservation. Again, we don't want the application to be locked for the wrong reason, e.g. lock held by a former process, but hostname has changed in the mean time. We also take the opportunity to use shorter timeout and delay betweeni locking attempts. The customatbility of the breaking logic and of the timeouts could be submitted UPSTREAM as options to Mercurial core, although it would be of limited interest to Heptapod right now: we cannot readily apply the same logic to other Mercurial locks, because HGitaly does not have exclusivity on them. Querying the system service manager (systemd or runit) about the Heptapod services would be a promising idea to do better, but we'd still have the issue that pushes over SSH would create standalone processes. diff --git a/hgitaly/tests/test_workdir.py b/hgitaly/tests/test_workdir.py --- 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,47 +316,62 @@ 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 """ + 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') - 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') -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() + 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)) - 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,10 +408,10 @@ 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') - 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 --- 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,7 +18,9 @@ from mercurial import ( cmdutil, commands, + error, hg, + lock, ) from mercurial.repoview import _filteredrepotypes @@ -132,8 +135,72 @@ 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 -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: