From 790c4e6eb22498b1a48dcc4655174dff65b2d9c7 Mon Sep 17 00:00:00 2001 From: Vaclav Petras Date: Mon, 28 Sep 2026 11:10:47 -0400 Subject: [PATCH] grass.pygrass: Fix double lock release when stopping RPC servers The forkserver start method test failed intermittently in CI, including on main, because the RPC servers of grass.pygrass and grass.temporal released their lock twice. The handler of the stop message, which the client sends to end the server process, released the lock inside a with-lock block, so the lock was released again when the block exited. The client terminated the server right after sending the message, so the resulting traceback appeared only when the server handled the message first. At exit, multiprocessing terminated the servers before grass.temporal stopped them. The checker thread then restarted a server without a session, the server failed in G_gisinit(), and its fatal error handler released a lock it did not hold. Remove the lock release from the stop and fatal error handlers. Wait for the server to exit after the stop message before terminating it. Import multiprocessing.util in grass.temporal before registering its exit function so that it runs before the exit handler of multiprocessing. Do not restart a server from the checker thread once stop was requested, and raise FatalError instead of restarting a server that multiprocessing terminated at exit, because the restarted server would never be stopped. Check for this only after the server was found dead, under the lock, which is race-free because multiprocessing sets its exiting flag before it terminates the servers. The failures were identified and the fix written with AI assistance (Claude Code with Opus 5.5, reviewed by Fable 5.1). --- python/grass/pygrass/rpc/__init__.py | 5 +--- python/grass/pygrass/rpc/base.py | 19 +++++++++++++- .../rpc/tests/grass_pygrass_rpc_base_test.py | 25 +++++++++++++++++++ .../grass/temporal/c_libraries_interface.py | 5 +--- python/grass/temporal/core.py | 5 +++- 5 files changed, 49 insertions(+), 10 deletions(-) create mode 100644 python/grass/pygrass/rpc/tests/grass_pygrass_rpc_base_test.py 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)