[pull] main from django:main

AUTHORS

@@ -34,6 +34,7 @@ answer newbie questions, and generally made Django that much better:
     Ahmad Alhashemi <trans@ahmadh.com>
     Ahmad Al-Ibrahim
     Ahmed Eltawela <https://github.com/ahmedabt>
+    Ahmed Mohamed <https://github.com/ahmed5145>
     Ahmed Nassar <https://ahmednassar7.github.io/>
     Aina Anjary Fenomamy R. <anja1catus@gmail.com>
     ajs <adi@sieker.info>
@@ -688,6 +689,7 @@ answer newbie questions, and generally made Django that much better:
     Marc-Aurèle Brothier <ma.brothier@gmail.com>
     Margaret Fero <maggiefero@gmail.com>
     Marian Andre <django@andre.sk>
+    Mariano Baragiola <mbaragiola@linux.com>
     Marijke Luttekes <mail@marijkeluttekes.dev>
     Marijn Vriens <marijn@metronomo.cl>
     Mario Gonzalez <gonzalemario@gmail.com>

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))

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

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):

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):

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):