mirror of
https://github.com/ansible/awx.git
synced 2026-10-29 22:03:20 -02:30
AAP-87883: Use shared lock for project copy when no sync is needed (#16587)
Use shared lock for project copy when no sync is needed When a job template runs and no project sync is required (the common case for slice jobs), multiple jobs were serializing behind an exclusive lock even though they were only reading the shared project directory to copy it. Introduce read/write lock semantics using fcntl LOCK_SH/LOCK_EX: - acquire_lock gains an exclusive= parameter (default True, preserving existing behavior for all callers that don't pass it) - sync_and_copy always acquires LOCK_SH first, then refreshes DB state and calls get_sync_needs() under the lock — preserving the invariant that the sync check always happens under a lock (no TOCTOU window) - If sync is needed, LOCK_SH is released and LOCK_EX is acquired before proceeding; sync_and_copy_without_lock re-checks internally - project.refresh_from_db() is guarded by project.pk to handle unsaved model instances used in unit tests This allows slice jobs backed by the same project to copy the project tree concurrently instead of sequentially. Measured improvement on a 200MB/54k-file project with 8 slice jobs: 140.5s total lock wait → 0s, wall time 42.6s → 16.9s (2.5x). Assisted-by: Claude Code via Google Vertex AI <cmeyers+claude@redhat.com>
This commit is contained in:
Chris Meyers
+35
-3
@@ -547,7 +547,13 @@ class BaseTask(object):
|
||||
os.close(self.lock_fd)
|
||||
self.lock_fd = None
|
||||
|
||||
def acquire_lock(self, project, unified_job_id=None):
|
||||
def acquire_lock(self, project, unified_job_id=None, exclusive=True):
|
||||
"""Acquire a file lock on the project's local source tree.
|
||||
|
||||
Uses LOCK_EX (exclusive) when the tree will be modified, or LOCK_SH (shared)
|
||||
when only reading (e.g. copying). Polls until the lock is granted, checking
|
||||
for cancellation on each iteration.
|
||||
"""
|
||||
if not os.path.exists(settings.PROJECTS_ROOT):
|
||||
os.mkdir(settings.PROJECTS_ROOT)
|
||||
|
||||
@@ -565,11 +571,12 @@ class BaseTask(object):
|
||||
logger.error("I/O error({0}) while trying to open lock file [{1}]: {2}".format(e.errno, lock_path, e.strerror))
|
||||
raise
|
||||
|
||||
lock_type = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH
|
||||
emitted_lockfile_log = False
|
||||
start_time = time.time()
|
||||
while True:
|
||||
try:
|
||||
fcntl.lockf(self.lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
fcntl.lockf(self.lock_fd, lock_type | fcntl.LOCK_NB)
|
||||
break
|
||||
except IOError as e:
|
||||
if e.errno not in (errno.EAGAIN, errno.EACCES):
|
||||
@@ -958,10 +965,35 @@ class SourceControlMixin(BaseTask):
|
||||
RunProjectUpdate.make_local_copy(project, private_data_dir)
|
||||
|
||||
def sync_and_copy(self, project, private_data_dir, scm_branch=None):
|
||||
self.acquire_lock(project, self.instance.id)
|
||||
"""Copy project content to private_data_dir, syncing from SCM only if needed.
|
||||
|
||||
Acquires a shared lock first so concurrent copy-only jobs (e.g. slice jobs) can
|
||||
run in parallel. Upgrades to an exclusive lock only when the project tree needs
|
||||
to be modified (fresh clone, revision mismatch, or missing cache). DB state is
|
||||
refreshed after each lock acquisition to account for concurrent updates.
|
||||
"""
|
||||
# Always start with a shared lock so concurrent copy-only jobs don't serialize.
|
||||
# LOCK_SH waits for any in-flight LOCK_EX to drain, making the tree stable.
|
||||
# Refresh DB state after acquiring so get_sync_needs sees the current revision,
|
||||
# then upgrade to LOCK_EX only if a sync is actually required.
|
||||
self.acquire_lock(project, self.instance.id, exclusive=False)
|
||||
is_commit = False
|
||||
try:
|
||||
original_branch = None
|
||||
if project.pk:
|
||||
project.refresh_from_db()
|
||||
sync_needs = self.get_sync_needs(project, scm_branch=scm_branch)
|
||||
if sync_needs:
|
||||
# Tree needs modification — upgrade to exclusive.
|
||||
# POSIX advisory locks cannot be upgraded atomically: LOCK_SH must be
|
||||
# released before LOCK_EX can be granted, leaving a window where another
|
||||
# process may sync the project. Refresh after re-acquiring so
|
||||
# sync_and_copy_without_lock operates on current DB state.
|
||||
self.release_lock(project)
|
||||
self.acquire_lock(project, self.instance.id, exclusive=True)
|
||||
if project.pk:
|
||||
project.refresh_from_db()
|
||||
|
||||
failed_reason = project.get_reason_if_failed()
|
||||
if failed_reason:
|
||||
self.update_model(self.instance.pk, status='failed', job_explanation=failed_reason)
|
||||
|
||||
@@ -1636,6 +1636,100 @@ def test_acquire_lock_acquisition_fail_logged(fcntl_lockf, logger_mock, os_close
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'kwargs,expected_flag',
|
||||
[
|
||||
({}, fcntl.LOCK_EX | fcntl.LOCK_NB), # default is exclusive
|
||||
({'exclusive': False}, fcntl.LOCK_SH | fcntl.LOCK_NB),
|
||||
],
|
||||
)
|
||||
@mock.patch('os.open')
|
||||
@mock.patch('fcntl.lockf')
|
||||
def test_acquire_lock_flag(fcntl_lockf, os_open, mock_me, kwargs, expected_flag):
|
||||
"""acquire_lock passes the correct fcntl flag based on the exclusive parameter."""
|
||||
instance = mock.Mock()
|
||||
instance.get_lock_file.return_value = '/fake/path.lock'
|
||||
os_open.return_value = 3
|
||||
|
||||
task = jobs.RunProjectUpdate()
|
||||
task.acquire_lock(instance, **kwargs)
|
||||
|
||||
assert fcntl_lockf.call_args[0][1] == expected_flag
|
||||
|
||||
|
||||
# Each entry: (sync_needs, has_pk, expected exclusive= sequence, expected refresh_from_db count)
|
||||
# - copy-only path: refresh once under LOCK_SH
|
||||
# - upgrade path: refresh under LOCK_SH and again under LOCK_EX (POSIX locks cannot
|
||||
# be upgraded atomically, so state may change in the gap between releasing SH and
|
||||
# acquiring EX)
|
||||
# - pk=None cases: refresh_from_db is skipped (unsaved model instances used in some tests)
|
||||
@pytest.mark.parametrize(
|
||||
'sync_needs,has_pk,expected_exclusive_seq,expected_refresh_count',
|
||||
[
|
||||
([], True, [False], 1), # copy-only: shared lock, refresh once
|
||||
(['update_git'], True, [False, True], 2), # upgrade path: refresh under SH and EX
|
||||
([], False, [False], 0), # copy-only, no pk: refresh skipped
|
||||
(['update_git'], False, [False, True], 0), # upgrade path, no pk: refresh skipped
|
||||
],
|
||||
)
|
||||
@mock.patch('awx.main.tasks.jobs.SourceControlMixin.release_lock')
|
||||
@mock.patch('awx.main.tasks.jobs.SourceControlMixin.acquire_lock')
|
||||
@mock.patch('awx.main.tasks.jobs.SourceControlMixin.sync_and_copy_without_lock')
|
||||
@mock.patch('awx.main.tasks.jobs.SourceControlMixin.get_sync_needs')
|
||||
def test_sync_and_copy_lock_behavior(
|
||||
get_sync_needs, sync_without_lock, acquire_lock, release_lock, mock_me, sync_needs, has_pk, expected_exclusive_seq, expected_refresh_count
|
||||
):
|
||||
"""sync_and_copy acquires LOCK_SH first, upgrading to LOCK_EX only when a sync is needed.
|
||||
project.refresh_from_db() is called after each lock acquisition to ensure current DB state,
|
||||
but only when the project has a pk (i.e. is persisted to the database)."""
|
||||
get_sync_needs.return_value = sync_needs
|
||||
|
||||
project = mock.Mock()
|
||||
project.pk = 1 if has_pk else None
|
||||
project.get_reason_if_failed.return_value = None
|
||||
project.scm_type = 'git'
|
||||
project.scm_branch = 'main'
|
||||
|
||||
task = jobs.RunJob()
|
||||
task.instance = mock.Mock()
|
||||
task.instance.id = 1
|
||||
task.update_model = mock.Mock(return_value=task.instance)
|
||||
|
||||
task.sync_and_copy(project, '/fake/private_data_dir')
|
||||
|
||||
assert acquire_lock.call_count == len(expected_exclusive_seq)
|
||||
for i, exclusive in enumerate(expected_exclusive_seq):
|
||||
assert acquire_lock.call_args_list[i] == mock.call(project, task.instance.id, exclusive=exclusive)
|
||||
assert release_lock.call_count == len(expected_exclusive_seq)
|
||||
assert project.refresh_from_db.call_count == expected_refresh_count
|
||||
|
||||
|
||||
@mock.patch('awx.main.tasks.jobs.SourceControlMixin.release_lock')
|
||||
@mock.patch('awx.main.tasks.jobs.SourceControlMixin.acquire_lock')
|
||||
@mock.patch('awx.main.tasks.jobs.SourceControlMixin.sync_and_copy_without_lock')
|
||||
@mock.patch('awx.main.tasks.jobs.SourceControlMixin.get_sync_needs')
|
||||
def test_sync_and_copy_get_sync_needs_called_under_shared_lock(get_sync_needs, sync_without_lock, acquire_lock, release_lock, mock_me):
|
||||
"""get_sync_needs is called after acquiring LOCK_SH, not before any lock."""
|
||||
call_order = []
|
||||
acquire_lock.side_effect = lambda *a, **kw: call_order.append('acquire')
|
||||
get_sync_needs.side_effect = lambda *a, **kw: call_order.append('get_sync_needs') or []
|
||||
|
||||
project = mock.Mock()
|
||||
project.get_reason_if_failed.return_value = None
|
||||
project.scm_type = 'git'
|
||||
project.scm_branch = 'main'
|
||||
|
||||
task = jobs.RunJob()
|
||||
task.instance = mock.Mock()
|
||||
task.instance.id = 1
|
||||
task.update_model = mock.Mock(return_value=task.instance)
|
||||
|
||||
task.sync_and_copy(project, '/fake/private_data_dir')
|
||||
|
||||
assert call_order[0] == 'acquire', 'lock must be acquired before get_sync_needs is called'
|
||||
assert call_order[1] == 'get_sync_needs'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('injector_cls', [cls for cls in ManagedCredentialType.registry.values() if cls.injectors])
|
||||
def test_managed_injector_redaction(injector_cls):
|
||||
"""See awx.main.models.inventory.PluginFileInjector._get_shared_env
|
||||
|
||||
Reference in New Issue
Block a user