diff --git a/python/grass/pygrass/rpc/__init__.py b/python/grass/pygrass/rpc/__init__.py index 839bad1d509..eb21376ba2b 100644 --- a/python/grass/pygrass/rpc/__init__.py +++ b/python/grass/pygrass/rpc/__init__.py @@ -185,7 +185,6 @@ def _fatal_error(lock, conn, data): def _stop(lock, conn, data): conn.close() - lock.release() sys.exit() @@ -205,11 +204,9 @@ def error_handler(data): """This function will be called in case of a fatal error in libgis""" # sys.stderr.write("Error handler was called\n") # We send an exception that will be handled in - # the parent process, then close the pipe - # and release any possible lock + # the parent process, then close the pipe. conn.send(FatalError("G_fatal_error() was called in the server process")) conn.close() - lock.release() CALLBACK = CFUNCTYPE(c_void_p, c_void_p) CALLBACK.restype = c_void_p diff --git a/python/grass/pygrass/rpc/base.py b/python/grass/pygrass/rpc/base.py index 4a754790b87..b88b61400f8 100644 --- a/python/grass/pygrass/rpc/base.py +++ b/python/grass/pygrass/rpc/base.py @@ -11,6 +11,7 @@ from __future__ import annotations import logging +import multiprocessing.util import sys import threading import time @@ -26,6 +27,10 @@ logger: logging.Logger = logging.getLogger(__name__) +# The function is not documented (although multiprocessing.queues uses it), +# so do not fail if it is missing. +_mp_is_exiting = getattr(multiprocessing.util, "is_exiting", lambda: False) + ############################################################################### @@ -131,10 +136,16 @@ def thread_checker(self): """Check every 200 micro seconds if the server process is alive""" while True: time.sleep(0.2) - self._check_restart_server(caller="Server check thread") + # Check before restarting so that a server is not restarted + # once stop was requested. with self.threadLock: if self.stopThread is True: return + try: + self._check_restart_server(caller="Server check thread") + except FatalError: + # Raised only when the server was terminated at exit. + return def start_server(self): """This function must be re-implemented in the subclasses""" @@ -159,6 +170,10 @@ def _check_restart_server(self, caller="main thread") -> None: with self.threadLock: if self.server is not None and self.server.is_alive() is True: return + # Do not restart a server that multiprocessing terminated at exit. + if _mp_is_exiting(): + msg = f"The libgis server was terminated at exit, caller: {caller}" + raise FatalError(msg) if self.client_conn is not None: self.client_conn.close() if self.server_conn is not None: @@ -206,6 +221,8 @@ def stop(self): 0, ] ) + # Let the server exit on its own before falling back to terminate. + self.server.join(timeout=5) self.server.terminate() # A process still starting up reopens the lock semaphore by name, # so it must be gone before the reference is dropped below. diff --git a/python/grass/pygrass/rpc/tests/grass_pygrass_rpc_base_test.py b/python/grass/pygrass/rpc/tests/grass_pygrass_rpc_base_test.py new file mode 100644 index 00000000000..75866143202 --- /dev/null +++ b/python/grass/pygrass/rpc/tests/grass_pygrass_rpc_base_test.py @@ -0,0 +1,25 @@ +# Copyright (C) 2026 by the GRASS Development Team +# SPDX-License-Identifier: GPL-2.0-or-later + +"""Test the RPC server base class""" + +import pytest + +from grass.exceptions import FatalError +from grass.pygrass.rpc import base + + +def test_no_restart_at_exit(monkeypatch): + """A server terminated at exit is not restarted and the check raises""" + provider = base.RPCServerBase() + try: + monkeypatch.setattr(base, "_mp_is_exiting", lambda: True) + server = provider.server + server.terminate() + server.join() + with pytest.raises(FatalError, match="terminated at exit"): + provider.check_server() + assert provider.server is server + assert not provider.is_server_alive() + finally: + provider.stop() diff --git a/python/grass/temporal/c_libraries_interface.py b/python/grass/temporal/c_libraries_interface.py index 038d595649e..3804296b9d3 100644 --- a/python/grass/temporal/c_libraries_interface.py +++ b/python/grass/temporal/c_libraries_interface.py @@ -1176,7 +1176,6 @@ def _convert_timestamp_from_grass(ts): def _stop(lock: _LockLike, conn: Connection, data) -> None: libgis.G_debug(1, "Stop C-interface server") conn.close() - lock.release() sys.exit() @@ -1196,11 +1195,9 @@ def error_handler(data) -> None: """This function will be called in case of a fatal error in libgis""" # sys.stderr.write("Error handler was called\n") # We send an exception that will be handled in - # the parent process, then close the pipe - # and release any possible lock + # the parent process, then close the pipe. conn.send(FatalError()) conn.close() - lock.release() CALLBACK = CFUNCTYPE(c_void_p, c_void_p) CALLBACK.restype = c_void_p diff --git a/python/grass/temporal/core.py b/python/grass/temporal/core.py index 4ac181a3f81..1dbb8e3b558 100644 --- a/python/grass/temporal/core.py +++ b/python/grass/temporal/core.py @@ -31,6 +31,7 @@ from __future__ import annotations import atexit +import multiprocessing.util # noqa: F401 import os import sqlite3 from datetime import datetime @@ -491,7 +492,9 @@ def stop_subprocesses() -> None: c_library_interface.stop() -# We register this function to be called at exit +# Importing multiprocessing.util registers the multiprocessing exit handler, +# so this function, registered later, runs before it (atexit is LIFO) and stops +# the servers before multiprocessing terminates them. atexit.register(stop_subprocesses)