Fix Python process-pool test server lifecycle

Client: py

Co-Authored-By: OpenAI Codex (GPT-5.6) <codex@openai.com>

Author: kpumuk

test/py/RunClientServer.py

@@ -47,7 +47,6 @@
 FRAMED = ["TNonblockingServer"]
 SKIP_ZLIB = ['TNonblockingServer', 'THttpServer']
 SKIP_SSL = ['THttpServer']
-EXTRA_DELAY = dict(TProcessPoolServer=5.5)
 
 PROTOS = [
     'accel',
@@ -78,6 +77,49 @@ def relfile(fname):
     return os.path.join(SCRIPT_DIR, fname)
 
 
+def terminate_process_group(process, timeout=5):
+    """Terminate a test server, killing it if graceful shutdown times out."""
+    if platform.system() == 'Windows':
+        if process.poll() is not None:
+            return
+        process.terminate()
+        try:
+            process.wait(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait()
+        return
+
+    process_group = process.pid
+    try:
+        os.killpg(process_group, signal.SIGTERM)
+    except ProcessLookupError:
+        process.wait()
+        return
+
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        process.poll()
+        try:
+            os.killpg(process_group, 0)
+        except ProcessLookupError:
+            process.wait()
+            return
+        except PermissionError:
+            # macOS can report EPERM briefly while group members are exiting.
+            pass
+        time.sleep(0.01)
+
+    try:
+        os.killpg(process_group, signal.SIGKILL)
+    except ProcessLookupError:
+        pass
+    except PermissionError:
+        if process.poll() is None:
+            process.kill()
+    process.wait()
+
+
 def setup_pypath(libdir, gendir):
     dirs = [libdir, gendir]
     env = copy.deepcopy(os.environ)
@@ -182,22 +224,7 @@ def ensureServerAlive():
                 ensureServerAlive()
             except Exception as exc:
                 cleanup_exc = exc
-            extra_sleep = EXTRA_DELAY.get(server_class, 0)
-            if extra_sleep > 0 and verbose > 0:
-                print('Giving %s (proto=%s,zlib=%s,ssl=%s) an extra %d seconds for child'
-                      'processes to terminate via alarm'
-                      % (server_class, proto, use_zlib, use_ssl, extra_sleep))
-                time.sleep(extra_sleep)
-            sig = signal.SIGKILL if platform.system() != 'Windows' else signal.SIGABRT
-            try:
-                if platform.system() == 'Windows':
-                    os.kill(serverproc.pid, sig)
-                else:
-                    # POSIX: kill the whole process group to reap forked children.
-                    os.killpg(serverproc.pid, sig)
-            except OSError:
-                pass
-            serverproc.wait()
+            terminate_process_group(serverproc)
         try:
             os.unlink(port_file)
         except OSError:

test/py/TestServer.py

@@ -413,28 +413,14 @@ def main(options):
     if server_type == "TNonblockingServer":
         server = TNonblockingServer.TNonblockingServer(processor, transport, inputProtocolFactory=pfactory)
     elif server_type == "TProcessPoolServer":
-        import signal
         from thrift.server import TProcessPoolServer
         server = TProcessPoolServer.TProcessPoolServer(processor, transport, tfactory, pfactory)
         server.setNumWorkers(5)
 
-        def set_alarm():
-            def clean_shutdown(signum, frame):
-                for worker in server.workers:
-                    if options.verbose > 0:
-                        logging.info('Terminating worker: %s' % worker)
-                    worker.terminate()
-                if options.verbose > 0:
-                    logging.info('Shutting down server')
-                # server.stop() would deadlock here: it calls Condition.notify()
-                # reentrantly on the same thread that's blocked in serve()'s
-                # Condition.wait(), which can never post the wake ack notify()
-                # is waiting for (THRIFT-6082). Workers are already terminated,
-                # so just exit directly.
-                os._exit(0)
-            signal.signal(signal.SIGALRM, clean_shutdown)
-            signal.alarm(4)
-        set_alarm()
+        def stop_process_pool(signum, frame):
+            raise SystemExit
+
+        signal.signal(signal.SIGTERM, stop_process_pool)
     else:
         # look up server class dynamically to instantiate server
         ServerClass = getattr(TServer, server_type)