From b84931a49b8ee4e88c9eb44ad444637e75aad3af Mon Sep 17 00:00:00 2001 From: Chris Meyers Date: Thu, 20 Aug 2026 09:10:29 -0400 Subject: [PATCH] AAP-87883: Use shared lock for project copy when no sync is needed (#16587) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- awx/main/tasks/jobs.py | 38 ++++++++++++- awx/main/tests/unit/test_tasks.py | 94 +++++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+), 3 deletions(-) diff --git a/awx/main/tasks/jobs.py b/awx/main/tasks/jobs.py index 1b47de4f45..5fb06bf87a 100644 --- a/awx/main/tasks/jobs.py +++ b/awx/main/tasks/jobs.py @@ -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) diff --git a/awx/main/tests/unit/test_tasks.py b/awx/main/tests/unit/test_tasks.py index 68b4d8e6d0..ba490e7252 100644 --- a/awx/main/tests/unit/test_tasks.py +++ b/awx/main/tests/unit/test_tasks.py @@ -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