diff --git a/AUTHORS b/AUTHORS index eeb2e7acbbc9..26b8d73c5517 100644 --- a/AUTHORS +++ b/AUTHORS @@ -34,6 +34,7 @@ answer newbie questions, and generally made Django that much better: Ahmad Alhashemi Ahmad Al-Ibrahim Ahmed Eltawela + Ahmed Mohamed Ahmed Nassar Aina Anjary Fenomamy R. ajs @@ -688,6 +689,7 @@ answer newbie questions, and generally made Django that much better: Marc-Aurèle Brothier Margaret Fero Marian Andre + Mariano Baragiola Marijke Luttekes Marijn Vriens Mario Gonzalez diff --git a/django/contrib/gis/db/backends/postgis/schema.py b/django/contrib/gis/db/backends/postgis/schema.py index 62577547b049..e77c0390acda 100644 --- a/django/contrib/gis/db/backends/postgis/schema.py +++ b/django/contrib/gis/db/backends/postgis/schema.py @@ -1,4 +1,4 @@ -from django.contrib.gis.db.models import GeometryField +from django.contrib.gis.db.models.fields import BaseSpatialField from django.db.backends.postgresql.schema import DatabaseSchemaEditor from django.db.models.expressions import Col, Func @@ -82,10 +82,10 @@ def _alter_field( ) old_field_spatial_index = ( - isinstance(old_field, GeometryField) and old_field.spatial_index + isinstance(old_field, BaseSpatialField) and old_field.spatial_index ) new_field_spatial_index = ( - isinstance(new_field, GeometryField) and new_field.spatial_index + isinstance(new_field, BaseSpatialField) and new_field.spatial_index ) if not old_field_spatial_index and new_field_spatial_index: self.execute(self._create_spatial_index_sql(model, new_field)) diff --git a/django/test/runner.py b/django/test/runner.py index 6df7b87fabd1..b5d8782726a2 100644 --- a/django/test/runner.py +++ b/django/test/runner.py @@ -12,6 +12,7 @@ import random import sys import textwrap +import time import unittest import unittest.suite from collections import defaultdict @@ -418,8 +419,59 @@ def parallel_type(value): _worker_id = 0 +def _claim_database_clone_slot(worker_slots, timeout=5): + """ + Claim a database clone slot (1-based index) for this worker process. + + Each slot of the shared worker_slots array stores the pid of the worker + process that owns it, or 0 if the slot is free. When multiprocessing.Pool + spawns a replacement for an exited worker, the replacement waits for the + parent process to release the dead worker's slot instead of being assigned + an index that has no matching database clone. + """ + pid = os.getpid() + deadline = time.monotonic() + timeout + while True: + with worker_slots.get_lock(): + free_index = None + for index, owner in enumerate(worker_slots): + # Prefer a slot this pid already owns over a free one, so that + # a pid reused by the OS for a replacement worker can never + # hold two slots at once (which would leave the exited + # worker's slot stuck until the replacement finishes). + if owner == pid: + return index + 1 + if owner == 0 and free_index is None: + free_index = index + if free_index is not None: + worker_slots[free_index] = pid + return free_index + 1 + if time.monotonic() > deadline: + raise RuntimeError( + f"Test worker (pid {pid}) could not acquire a test database " + f"clone within {timeout} seconds because all " + f"{len(worker_slots)} clones are assigned to running workers." + ) + time.sleep(0.05) + + +def _release_dead_worker_slots(worker_slots): + """ + Release database clone slots owned by worker processes that exited, so + that replacement workers spawned by multiprocessing.Pool can reuse them. + + This runs in the parent process, where pool workers are visible as active + children. + """ + with worker_slots.get_lock(): + alive_pids = {child.pid for child in multiprocessing.active_children()} + for index, owner in enumerate(worker_slots): + if owner and owner not in alive_pids: + worker_slots[index] = 0 + + def _init_worker( - counter, + worker_slots, initial_settings=None, serialized_contents=None, process_setup=None, @@ -436,9 +488,7 @@ def _init_worker( global _worker_id - with counter.get_lock(): - counter.value += 1 - _worker_id = counter.value + _worker_id = _claim_database_clone_slot(worker_slots) is_spawn_or_forkserver = multiprocessing.get_start_method() in { "forkserver", @@ -474,13 +524,11 @@ def _init_worker( ) -def _safe_init_worker(init_worker, counter, *args, **kwargs): +def _safe_init_worker(init_worker, init_failed, *args, **kwargs): try: - init_worker(counter, *args, **kwargs) + init_worker(*args, **kwargs) except Exception: - with counter.get_lock(): - # Set a value that will not increment above zero any time soon. - counter.value = -1000 + init_failed.value = True raise @@ -554,7 +602,9 @@ def run(self, result): exception classes which cannot be unpickled. """ self.initialize_suite() - counter = multiprocessing.Value(ctypes.c_int, 0) + # One slot per db clone, holding the pid of the worker process (or 0). + worker_slots = multiprocessing.Array(ctypes.c_longlong, self.processes) + init_failed = multiprocessing.Value(ctypes.c_bool, False) args = [ (self.runner_class, index, subsuite, self.failfast, self.buffer) for index, subsuite in enumerate(self.subsuites) @@ -566,7 +616,8 @@ def run(self, result): processes=self.processes, initializer=functools.partial(_safe_init_worker, self.init_worker.__func__), initargs=[ - counter, + init_failed, + worker_slots, self.initial_settings, self.serialized_contents, self.process_setup.__func__, @@ -582,10 +633,14 @@ def run(self, result): pool.terminate() break + # Return the database clones of exited workers to the pool of + # available slots so that replacement workers can reuse them. + _release_dead_worker_slots(worker_slots) + try: subsuite_index, events = test_results.next(timeout=0.1) except multiprocessing.TimeoutError as err: - if counter.value < 0: + if init_failed.value: err.add_note("ERROR: _init_worker failed, see prior traceback") raise continue diff --git a/tests/gis_tests/gis_migrations/test_operations.py b/tests/gis_tests/gis_migrations/test_operations.py index 98201ed3f782..7a1b5a483f22 100644 --- a/tests/gis_tests/gis_migrations/test_operations.py +++ b/tests/gis_tests/gis_migrations/test_operations.py @@ -292,6 +292,45 @@ def test_alter_field_remove_spatial_index(self): ) self.assertSpatialIndexNotExists("gis_neighborhood", "geom") + @skipUnlessDBFeature("can_alter_geometry_field", "supports_raster") + def test_alter_raster_field_add_spatial_index(self): + if not self.has_spatial_indexes: + self.skipTest("No support for Spatial indexes") + + self.alter_gis_model( + migrations.AddField, + "Neighborhood", + "heatmap", + fields.RasterField, + field_class_kwargs={"spatial_index": False}, + ) + self.assertSpatialIndexNotExists("gis_neighborhood", "heatmap", raster=True) + + self.alter_gis_model( + migrations.AlterField, + "Neighborhood", + "heatmap", + fields.RasterField, + field_class_kwargs={"spatial_index": True}, + ) + self.assertSpatialIndexExists("gis_neighborhood", "heatmap", raster=True) + + @skipUnlessDBFeature("can_alter_geometry_field", "supports_raster") + def test_alter_raster_field_remove_spatial_index(self): + if not self.has_spatial_indexes: + self.skipTest("No support for Spatial indexes") + + self.assertSpatialIndexExists("gis_neighborhood", "rast", raster=True) + + self.alter_gis_model( + migrations.AlterField, + "Neighborhood", + "rast", + fields.RasterField, + field_class_kwargs={"spatial_index": False}, + ) + self.assertSpatialIndexNotExists("gis_neighborhood", "rast", raster=True) + @skipUnlessDBFeature("can_alter_geometry_field") @skipUnless(connection.vendor == "mysql", "MySQL specific test") def test_alter_field_nullable_with_spatial_index(self): diff --git a/tests/test_runner/test_parallel.py b/tests/test_runner/test_parallel.py index d15843c097e4..df8532d4c612 100644 --- a/tests/test_runner/test_parallel.py +++ b/tests/test_runner/test_parallel.py @@ -1,12 +1,21 @@ +import ctypes +import multiprocessing +import os import pickle import sys import unittest +from unittest import mock from unittest.case import TestCase from unittest.result import TestResult from unittest.suite import TestSuite, _ErrorHolder from django.test import SimpleTestCase -from django.test.runner import ParallelTestSuite, RemoteTestResult +from django.test.runner import ( + ParallelTestSuite, + RemoteTestResult, + _claim_database_clone_slot, + _release_dead_worker_slots, +) from . import models @@ -226,6 +235,59 @@ def test_add_duration(self): self.assertEqual(result.collectedDurations, [("None", 2.3)]) +class DatabaseCloneSlotTests(SimpleTestCase): + def make_slots(self, *owners): + slots = multiprocessing.Array(ctypes.c_longlong, len(owners)) + for index, owner in enumerate(owners): + slots[index] = owner + return slots + + def test_claims_first_free_slot(self): + slots = self.make_slots(101, 0, 0) + self.assertEqual(_claim_database_clone_slot(slots), 2) + self.assertEqual(slots[1], os.getpid()) + + def test_reclaims_slot_owned_by_own_pid(self): + # The OS may reuse the pid of an exited worker whose slot was not + # released yet. + slots = self.make_slots(101, os.getpid(), 0) + self.assertEqual(_claim_database_clone_slot(slots), 2) + + def test_prefers_own_slot_over_earlier_free_slot(self): + # A pid already owning a slot must reclaim it rather than take an + # earlier free slot, otherwise a pid reused by the OS could hold two + # slots at once and strand the old one. + slots = self.make_slots(0, os.getpid()) + self.assertEqual(_claim_database_clone_slot(slots), 2) + # The earlier free slot is left untouched for another worker. + self.assertEqual(slots[0], 0) + + def test_times_out_when_no_slot_is_free(self): + slots = self.make_slots(101, 102) + msg = "could not acquire a test database clone" + with self.assertRaisesMessage(RuntimeError, msg): + _claim_database_clone_slot(slots, timeout=0) + + def test_release_dead_worker_slots(self): + alive_worker = mock.Mock(pid=103) + slots = self.make_slots(101, 0, 103) + with mock.patch.object( + multiprocessing, "active_children", return_value=[alive_worker] + ): + _release_dead_worker_slots(slots) + self.assertEqual(list(slots), [0, 0, 103]) + + def test_released_slot_is_claimed_by_replacement_worker(self): + dead_worker_pid = 101 + slots = self.make_slots(dead_worker_pid, 102) + with mock.patch.object( + multiprocessing, "active_children", return_value=[mock.Mock(pid=102)] + ): + _release_dead_worker_slots(slots) + self.assertEqual(_claim_database_clone_slot(slots), 1) + self.assertEqual(slots[0], os.getpid()) + + class ParallelTestSuiteTest(SimpleTestCase): @unittest.skipUnless(tblib is not None, "requires tblib to be installed") def test_handle_add_error_before_first_test(self): diff --git a/tests/test_runner/tests.py b/tests/test_runner/tests.py index 9173fa5d36a9..83843e3913f0 100644 --- a/tests/test_runner/tests.py +++ b/tests/test_runner/tests.py @@ -744,9 +744,9 @@ def run_suite(self, suite, **kwargs): self.assertIsInstance(initializer, functools.partial) self.assertIs(initializer.args[0], _init_worker) initargs = mocked_pool.call_args.kwargs["initargs"] - self.assertEqual(len(initargs), 7) - self.assertEqual(initargs[5], True) # debug_mode - self.assertEqual(initargs[6], {db.DEFAULT_DB_ALIAS}) # Used database aliases. + self.assertEqual(len(initargs), 8) + self.assertEqual(initargs[6], True) # debug_mode + self.assertEqual(initargs[7], {db.DEFAULT_DB_ALIAS}) # Used database aliases. class Ticket17477RegressionTests(AdminScriptTestCase):