diff --git a/CHANGES/3322.bugfix b/CHANGES/3322.bugfix new file mode 100644 index 00000000000..a6f1876cf2e --- /dev/null +++ b/CHANGES/3322.bugfix @@ -0,0 +1 @@ +Reduced lock contention for distribution updates that leave `base_path` unchanged. diff --git a/CHANGES/7896.bugfix b/CHANGES/7896.bugfix new file mode 100644 index 00000000000..a6f1876cf2e --- /dev/null +++ b/CHANGES/7896.bugfix @@ -0,0 +1 @@ +Reduced lock contention for distribution updates that leave `base_path` unchanged. diff --git a/pulpcore/app/viewsets/publication.py b/pulpcore/app/viewsets/publication.py index ae7e38a8e6d..d2bc7d79063 100644 --- a/pulpcore/app/viewsets/publication.py +++ b/pulpcore/app/viewsets/publication.py @@ -2,8 +2,11 @@ from django.db.models import Prefetch from django_filters import Filter +from drf_spectacular.utils import extend_schema from rest_framework import mixins, serializers +from rest_framework.response import Response +from pulpcore.app import tasks from pulpcore.app.models import ( ArtifactDistribution, ContentGuard, @@ -15,8 +18,10 @@ Repository, ) from pulpcore.app.models.publication import CompositeContentGuard +from pulpcore.app.response import OperationPostponedResponse from pulpcore.app.serializers import ( ArtifactDistributionSerializer, + AsyncOperationResponseSerializer, ContentGuardSerializer, ContentRedirectContentGuardSerializer, DistributionSerializer, @@ -43,6 +48,8 @@ WithContentInFilter, ) from pulpcore.filters import BaseFilterSet +from pulpcore.openapi import InheritSerializer +from pulpcore.tasking.tasks import dispatch class RepositoryThroughVersionFilter(Filter): @@ -527,7 +534,31 @@ def get_queryset(self): return qs def async_reserved_resources(self, instance): - """Return resource that locks all Distributions.""" + """ + Reserve safe distribution locks for async operations. + + The explicit distribution.base_path lock protects the domain-wide base_path invariant. + The older domain-scoped distributions lock remains shared so tasks queued before an upgrade + still overlap safely with new tasks. + """ + distribution_base_path = f"pdrn:{get_domain().pulp_id}:distribution.base_path" + if instance is None: + return [distribution_base_path] + + if getattr(self, "action", "") == "destroy": + return [instance] + + request_data = getattr(getattr(self, "request", None), "data", {}) + requested_base_path = request_data.get("base_path", instance.base_path) + if requested_base_path == instance.base_path: + return [instance] + + return [instance, distribution_base_path] + + def async_shared_resources(self, instance): + """ + Keep the legacy domain-scoped distribution lock shared for upgrade compatibility. + """ return [f"pdrn:{get_domain().pulp_id}:distributions"] @@ -567,11 +598,75 @@ class DistributionViewSet( LabelsMixin, ): """ - Provides read and list methods and also provides asynchronous CUD methods to dispatch tasks - with reservation that lock all Distributions preventing race conditions during base_path - checking. + Provides read and list methods plus asynchronous CUD methods that reserve per-distribution + locks, an explicit base_path lock when needed, and the legacy domain-wide distributions lock in + shared mode for upgrade compatibility. """ + @extend_schema( + description="Trigger an asynchronous create task", + responses={202: AsyncOperationResponseSerializer}, + ) + def create(self, request, *args, **kwargs): + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + app_label = self.queryset.model._meta.app_label + task = dispatch( + tasks.base.general_create, + exclusive_resources=self.async_reserved_resources(None), + shared_resources=self.async_shared_resources(None), + args=(app_label, serializer.__class__.__name__), + kwargs={"data": request.data}, + ) + return OperationPostponedResponse(task, request) + + @extend_schema( + description="Update the entity and trigger an asynchronous task if necessary", + responses={200: InheritSerializer, 202: AsyncOperationResponseSerializer}, + ) + def update(self, request, pk, **kwargs): + partial = kwargs.pop("partial", False) + instance = self.get_object() + serializer = self.get_serializer(instance, data=request.data, partial=partial) + serializer.is_valid(raise_exception=True) + + if all(getattr(instance, key) == value for key, value in serializer.validated_data.items()): + return Response(serializer.data) + + task = dispatch( + tasks.base.ageneral_update, + exclusive_resources=self.async_reserved_resources(instance), + shared_resources=self.async_shared_resources(instance), + args=(pk, instance._meta.app_label, serializer.__class__.__name__), + kwargs={"data": request.data, "partial": partial}, + immediate=self.ALLOW_NON_BLOCKING_UPDATE, + ) + return OperationPostponedResponse(task, request) + + @extend_schema( + description="Update the entity partially and trigger an asynchronous task if necessary", + responses={200: InheritSerializer, 202: AsyncOperationResponseSerializer}, + ) + def partial_update(self, request, *args, **kwargs): + kwargs["partial"] = True + return self.update(request, *args, **kwargs) + + @extend_schema( + description="Trigger an asynchronous delete task", + responses={202: AsyncOperationResponseSerializer}, + ) + def destroy(self, request, pk, **kwargs): + instance = self.get_object() + serializer = self.get_serializer(instance) + task = dispatch( + tasks.base.ageneral_delete, + exclusive_resources=self.async_reserved_resources(instance), + shared_resources=self.async_shared_resources(instance), + args=(pk, instance._meta.app_label, serializer.__class__.__name__), + immediate=self.ALLOW_NON_BLOCKING_DELETE, + ) + return OperationPostponedResponse(task, request) + class ArtifactDistributionViewSet(ReadOnlyDistributionViewSet): """ diff --git a/pulpcore/tests/functional/api/using_plugin/test_distributions.py b/pulpcore/tests/functional/api/using_plugin/test_distributions.py index 1870c86e462..5d7f57c1e59 100644 --- a/pulpcore/tests/functional/api/using_plugin/test_distributions.py +++ b/pulpcore/tests/functional/api/using_plugin/test_distributions.py @@ -167,6 +167,81 @@ def test_distribution_base_path( assert json.loads(exc.value.body)["base_path"] is not None +@pytest.mark.parallel +def test_distribution_update_task_reservations( + file_bindings, + monitor_task, +): + def has_shared_distributions_lock(task): + return any( + resource.startswith("shared:") and resource.endswith(":distributions") + for resource in task.reserved_resources_record + ) + + def has_exclusive_distributions_lock(task): + return any( + not resource.startswith("shared:") and resource.endswith(":distributions") + for resource in task.reserved_resources_record + ) + + def has_base_path_lock(task): + return any( + not resource.startswith("shared:") and resource.endswith(":distribution.base_path") + for resource in task.reserved_resources_record + ) + + create_task = monitor_task( + file_bindings.DistributionsFileApi.create( + {"name": str(uuid4()), "base_path": str(uuid4())} + ).task + ) + assert has_base_path_lock(create_task) + assert has_shared_distributions_lock(create_task) + assert not has_exclusive_distributions_lock(create_task) + distribution = file_bindings.DistributionsFileApi.read(create_task.created_resources[0]) + assert distribution.prn not in create_task.reserved_resources_record + + no_base_path_update_task = monitor_task( + file_bindings.DistributionsFileApi.partial_update( + distribution.pulp_href, + {"name": str(uuid4())}, + ).task + ) + assert distribution.prn in no_base_path_update_task.reserved_resources_record + assert not has_base_path_lock(no_base_path_update_task) + assert has_shared_distributions_lock(no_base_path_update_task) + assert not has_exclusive_distributions_lock(no_base_path_update_task) + + unchanged_base_path_update_task = monitor_task( + file_bindings.DistributionsFileApi.partial_update( + distribution.pulp_href, + {"name": str(uuid4()), "base_path": distribution.base_path}, + ).task + ) + assert distribution.prn in unchanged_base_path_update_task.reserved_resources_record + assert not has_base_path_lock(unchanged_base_path_update_task) + assert has_shared_distributions_lock(unchanged_base_path_update_task) + assert not has_exclusive_distributions_lock(unchanged_base_path_update_task) + + base_path_update_task = monitor_task( + file_bindings.DistributionsFileApi.partial_update( + distribution.pulp_href, {"base_path": str(uuid4())} + ).task + ) + assert distribution.prn in base_path_update_task.reserved_resources_record + assert has_base_path_lock(base_path_update_task) + assert has_shared_distributions_lock(base_path_update_task) + assert not has_exclusive_distributions_lock(base_path_update_task) + + delete_task = monitor_task( + file_bindings.DistributionsFileApi.delete(distribution.pulp_href).task + ) + assert distribution.prn in delete_task.reserved_resources_record + assert not has_base_path_lock(delete_task) + assert has_shared_distributions_lock(delete_task) + assert not has_exclusive_distributions_lock(delete_task) + + @pytest.mark.parallel def test_distribution_filtering( file_bindings,