grass.pygrass: Pin fork start method for RPC servers on Python 3.14

python/grass/pygrass/messages/__init__.py

@@ -17,15 +17,16 @@
 from __future__ import annotations
 
 import sys
-from multiprocessing import Lock, Pipe, Process
 from typing import TYPE_CHECKING, Literal, NoReturn
 
 import grass.lib.gis as libgis
 from grass.exceptions import FatalError
+from grass.script.utils import _get_multiprocessing_context
 
 if TYPE_CHECKING:
     from multiprocessing.connection import Connection
     from multiprocessing.context import _LockLike
+    from multiprocessing.process import BaseProcess
 
     _MessagesLiteral = Literal[
         "INFO", "IMPORTANT", "VERBOSE", "WARNING", "ERROR", "FATAL"
@@ -177,21 +178,27 @@ class Messenger:
 
     client_conn: Connection
     server_conn: Connection
-    server: Process
+    server: BaseProcess
 
     def __init__(self, raise_on_error: bool = False) -> None:
         self.raise_on_error = raise_on_error
-        self.client_conn, self.server_conn = Pipe()
-        self.lock = Lock()
-        self.server = Process(target=message_server, args=(self.lock, self.server_conn))
+        ctx = _get_multiprocessing_context()
+        self.client_conn, self.server_conn = ctx.Pipe()
+        self.lock = ctx.Lock()
+        self.server = ctx.Process(
+            target=message_server, args=(self.lock, self.server_conn)
+        )
         self.server.daemon = True
         self.server.start()
 
     def start_server(self) -> None:
         """Start the messenger server and open the pipe"""
-        self.client_conn, self.server_conn = Pipe()
-        self.lock = Lock()
-        self.server = Process(target=message_server, args=(self.lock, self.server_conn))
+        ctx = _get_multiprocessing_context()
+        self.client_conn, self.server_conn = ctx.Pipe()
+        self.lock = ctx.Lock()
+        self.server = ctx.Process(
+            target=message_server, args=(self.lock, self.server_conn)
+        )
         self.server.daemon = True
         self.server.start()
 

python/grass/pygrass/rpc/__init__.py

@@ -12,7 +12,6 @@
 
 import sys
 from ctypes import CFUNCTYPE, c_void_p
-from multiprocessing import Lock, Pipe, Process
 
 import grass.lib.gis as libgis
 from grass.exceptions import FatalError
@@ -21,6 +20,7 @@
 from grass.pygrass.raster import RasterRow, raster2numpy_img
 from grass.pygrass.vector import VectorTopo
 from grass.pygrass.vector.basic import Bbox
+from grass.script.utils import _get_multiprocessing_context
 
 from .base import RPCServerBase
 
@@ -249,9 +249,10 @@ def __init__(self):
 
     def start_server(self):
         """This function must be re-implemented in the subclasses"""
-        self.client_conn, self.server_conn = Pipe(True)
-        self.lock = Lock()
-        self.server = Process(
+        ctx = _get_multiprocessing_context()
+        self.client_conn, self.server_conn = ctx.Pipe(True)
+        self.lock = ctx.Lock()
+        self.server = ctx.Process(
             target=data_provider_server, args=(self.lock, self.server_conn)
         )
         self.server.daemon = True

python/grass/pygrass/rpc/base.py

@@ -16,10 +16,10 @@
 import sys
 import threading
 import time
-from multiprocessing import Lock, Pipe, Process
 from typing import TYPE_CHECKING, NoReturn
 
 from grass.exceptions import FatalError
+from grass.script.utils import _get_multiprocessing_context
 
 if TYPE_CHECKING:
     from multiprocessing.connection import Connection
@@ -142,9 +142,12 @@ def start_server(self):
         """This function must be re-implemented in the subclasses"""
         logger.debug("Start the libgis server")
 
-        self.client_conn, self.server_conn = Pipe(True)
-        self.lock = Lock()
-        self.server = Process(target=dummy_server, args=(self.lock, self.server_conn))
+        ctx = _get_multiprocessing_context()
+        self.client_conn, self.server_conn = ctx.Pipe(True)
+        self.lock = ctx.Lock()
+        self.server = ctx.Process(
+            target=dummy_server, args=(self.lock, self.server_conn)
+        )
         self.server.daemon = True
         self.server.start()
 

python/grass/script/tests/utils_test.py

@@ -1,5 +1,8 @@
 """Test functions in grass.script.utils"""
 
+import multiprocessing
+import sys
+
 import pytest
 
 import grass.script as gs
@@ -65,3 +68,12 @@ def test_resolve_nprocs(monkeypatch):
     assert gs.resolve_nprocs(-10) == 1
     with pytest.raises(ValueError, match="invalid literal for int"):
         gs.resolve_nprocs("not-a-number")
+
+
+def test_get_multiprocessing_context():
+    """Fork is used where available and safe, the platform default elsewhere."""
+    ctx = gutils._get_multiprocessing_context()
+    if sys.platform != "darwin" and "fork" in multiprocessing.get_all_start_methods():
+        assert ctx.get_start_method() == "fork"
+    else:
+        assert ctx.get_start_method() == multiprocessing.get_start_method()

python/grass/script/utils.py

@@ -29,6 +29,7 @@
 import uuid
 import random
 import string
+import sys
 
 from pathlib import Path
 from typing import TYPE_CHECKING, AnyStr, TypeVar, cast, overload
@@ -37,6 +38,7 @@
 if TYPE_CHECKING:
     from _typeshed import FileDescriptorOrPath, StrOrBytesPath, StrPath
     from collections.abc import Callable
+    from multiprocessing.context import BaseContext
 
 
 # Type variables
@@ -127,6 +129,29 @@ def resolve_nprocs(nprocs: int | str) -> int:
     return max(1, available + n)
 
 
+def _get_multiprocessing_context() -> BaseContext:
+    """Return the multiprocessing context to use for GRASS worker processes.
+
+    Python 3.14 changed the default multiprocessing start method from "fork"
+    to "forkserver" (on platforms where "fork" is available, except macOS).
+    GRASS worker processes rely on "fork" semantics: they inherit the
+    initialized C library state and the session environment from the parent.
+    Use "fork" wherever it is available, i.e. the default before 3.14, except
+    on macOS, which has defaulted to "spawn" since Python 3.8 and where
+    "fork" is unsafe. Everywhere else (Windows) keep the platform default.
+
+    This is a workaround until GRASS supports the "spawn" and "forkserver"
+    start methods.
+    """
+    # Imported here because grass.script is imported by every Python tool
+    # and only a few callers need multiprocessing.
+    import multiprocessing
+
+    if sys.platform != "darwin" and "fork" in multiprocessing.get_all_start_methods():
+        return multiprocessing.get_context("fork")
+    return multiprocessing.get_context()
+
+
 def diff_files(
     filename_a: FileDescriptorOrPath, filename_b: FileDescriptorOrPath
 ) -> list[str]:

python/grass/temporal/c_libraries_interface.py

@@ -16,7 +16,6 @@
 import sys
 from ctypes import CFUNCTYPE, POINTER, byref, c_int, c_void_p, cast
 from datetime import datetime
-from multiprocessing import Lock, Pipe, Process
 from typing import TYPE_CHECKING, Any, Literal
 
 import grass.lib.date as libdate
@@ -30,7 +29,7 @@
 from grass.pygrass.rpc.base import RPCServerBase
 from grass.pygrass.utils import decode
 from grass.pygrass.vector import VectorTopo
-from grass.script.utils import encode
+from grass.script.utils import _get_multiprocessing_context, encode
 import grass.script as gs
 
 if TYPE_CHECKING:
@@ -1471,9 +1470,10 @@ def __init__(self) -> None:
         RPCServerBase.__init__(self)
 
     def start_server(self) -> None:
-        self.client_conn, self.server_conn = Pipe(True)
-        self.lock = Lock()
-        self.server = Process(
+        ctx = _get_multiprocessing_context()
+        self.client_conn, self.server_conn = ctx.Pipe(True)
+        self.lock = ctx.Lock()
+        self.server = ctx.Process(
             target=c_library_server, args=(self.lock, self.server_conn)
         )
         self.server.daemon = True

python/grass/temporal/extract.py

@@ -12,10 +12,10 @@
 
 import re
 import sys
-from multiprocessing import Process
 
 import grass.script as gs
 from grass.exceptions import CalledModuleError
+from grass.script.utils import _get_multiprocessing_context
 
 from .abstract_map_dataset import AbstractMapDataset
 
@@ -209,6 +209,7 @@ def extract_dataset(
 
     # Run the mapcalc expression
     if expression:
+        ctx = _get_multiprocessing_context()
         proc_count = 0
         proc_list = []
 
@@ -264,15 +265,15 @@ def extract_dataset(
             # Add process to the process list
             if type == "raster":
                 msgr.verbose(_('Applying r.mapcalc expression: "%s"') % expr)
-                proc_list.append(Process(target=run_mapcalc2d, args=(expr,)))
+                proc_list.append(ctx.Process(target=run_mapcalc2d, args=(expr,)))
             elif type == "raster3d":
                 msgr.verbose(_('Applying r3.mapcalc expression: "%s"') % expr)
-                proc_list.append(Process(target=run_mapcalc3d, args=(expr,)))
+                proc_list.append(ctx.Process(target=run_mapcalc3d, args=(expr,)))
             elif type == "vector":
                 msgr.verbose(_('Applying v.extract where statement: "%s"') % expression)
                 if row["layer"]:
                     proc_list.append(
-                        Process(
+                        ctx.Process(
                             target=run_vector_extraction,
                             args=(
                                 row["name"] + "@" + row["mapset"],
@@ -285,7 +286,7 @@ def extract_dataset(
                     )
                 else:
                     proc_list.append(
-                        Process(
+                        ctx.Process(
                             target=run_vector_extraction,
                             args=(
                                 row["name"] + "@" + row["mapset"],

python/grass/temporal/mapcalc.py

@@ -12,10 +12,10 @@
 import copy
 import sys
 from datetime import datetime
-from multiprocessing import Process
 
 import grass.script as gs
 from grass.exceptions import CalledModuleError
+from grass.script.utils import _get_multiprocessing_context
 
 from .core import (
     SQLDatabaseInterfaceConnection,
@@ -236,6 +236,7 @@ def dataset_mapcalculator(
         num = len(map_matrix[0])
 
         # Parallel processing
+        ctx = _get_multiprocessing_context()
         proc_list = []
         proc_count = 0
 
@@ -311,9 +312,9 @@ def dataset_mapcalculator(
 
             # Start the parallel r.mapcalc computation
             if type == "raster":
-                proc_list.append(Process(target=_run_mapcalc2d, args=(expr,)))
+                proc_list.append(ctx.Process(target=_run_mapcalc2d, args=(expr,)))
             else:
-                proc_list.append(Process(target=_run_mapcalc3d, args=(expr,)))
+                proc_list.append(ctx.Process(target=_run_mapcalc3d, args=(expr,)))
             proc_list[proc_count].start()
             proc_count += 1
 

python/grass/temporal/tests/grass_temporal_c_libraries_interface_start_method_test.py

@@ -0,0 +1,80 @@
+# Copyright (C) 2026 by the GRASS Development Team
+# SPDX-License-Identifier: GPL-2.0-or-later
+
+"""Test that GRASS server processes start when the default start method is not fork.
+
+Python 3.14 made forkserver the default multiprocessing start method on Linux,
+which broke the RPC servers in grass.pygrass and grass.temporal. The servers
+now request a fork context explicitly. This test forces forkserver as the
+default in a subprocess to reproduce the Python 3.14 situation on any version.
+"""
+
+import multiprocessing
+import subprocess
+import sys
+
+import pytest
+
+SCRIPT = """
+import multiprocessing
+import sys
+
+
+def main():
+    import grass.script as gs
+    import grass.temporal as tgis
+    from grass.pygrass.messages import Messenger
+    from grass.pygrass.rpc import DataProvider
+    from grass.pygrass.rpc.base import RPCServerBase
+
+    project = sys.argv[1]
+    gs.create_project(project)
+    # Legacy init without env so that os.environ is set for grass.temporal.
+    session = gs.setup.init(project)
+    gs.run_command("g.region", n=10, s=0, e=10, w=0, res=1)
+    gs.run_command("r.mapcalc", expression="a = 1")
+    mapset = gs.gisenv()["MAPSET"]
+
+    messenger = Messenger()
+    messenger.message("test")
+    messenger.stop()
+
+    server = RPCServerBase()
+    server.stop()
+
+    provider = DataProvider()
+    assert provider.get_raster_image_as_np("a").size > 0
+    provider.stop()
+
+    tgis.init()
+    ciface = tgis.CLibrariesInterface()
+    assert ciface.raster_map_exists("a", mapset)
+    ciface.stop()
+
+    session.finish()
+
+
+if __name__ == "__main__":
+    multiprocessing.set_start_method("forkserver")
+    main()
+"""
+
+
+@pytest.mark.skipif(
+    sys.platform == "darwin" or "fork" not in multiprocessing.get_all_start_methods(),
+    reason="Fork is only requested where it is available and not macOS",
+)
+def test_servers_start_with_forkserver_default(tmp_path):
+    """Servers work and exit cleanly when forkserver is the default start method."""
+    script = tmp_path / "script.py"
+    script.write_text(SCRIPT)
+    result = subprocess.run(
+        [sys.executable, str(script), str(tmp_path / "project")],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+    # A failing server check thread does not change the exit code.
+    assert "Traceback" not in result.stderr, result.stderr
+    assert "leaked semaphore" not in result.stderr, result.stderr