FIX: Embedding rpc server shutdown (#1088)

Co-authored-by: Dipin Nair <dkunhamb@chqmechlinux01.ansys.com>
Co-authored-by: pyansys-ci-bot <92810346+pyansys-ci-bot@users.noreply.github.com>

doc/changelog.d/1088.fixed.md

@@ -0,0 +1 @@
+Emedding rpc server shutdown

src/ansys/mechanical/core/embedding/rpc/client.py

@@ -232,6 +232,6 @@ def is_alive(self):
     def exit(self):
         """Shuts down the Mechanical instance."""
         print("Requesting server shutdown ...")
-        self.root.service_exit()
+        self.service_exit()
         self.connection.close()
         print("Disconnected from server")

src/ansys/mechanical/core/embedding/rpc/server.py

@@ -23,6 +23,8 @@
 
 import fnmatch
 import os
+import threading
+import time
 import typing
 
 import rpyc
@@ -211,8 +213,11 @@ def exposed_service_download(self, remote_path):
 
     def exposed_service_exit(self):
         """Exit the server."""
+        print("Shutting down server ...")
         self._backgroundapp.stop()
         self._backgroundapp = None
+        self._server.stop_async()
+        print("Server stopped")
 
 
 class MechanicalEmbeddedServer:
@@ -231,6 +236,7 @@ def __init__(
         self._background_app = BackgroundApp(version=version)
         self._service = service
         self._methods = methods
+        self._exit_thread: threading.Thread = None
         print("Initializing Mechanical ...")
 
         self._port = self.get_free_port(port)
@@ -241,6 +247,7 @@ def __init__(
 
         my_service = self._service(self._background_app, self._methods, self._impl)
         self._server = ThreadedServer(my_service, port=self._port)
+        my_service._server = self
 
     @staticmethod
     def get_free_port(port=None):
@@ -270,8 +277,28 @@ def start(self) -> None:
                 print("User interrupt!")
         finally:
             conn.close()"""
+        print("Server exited!")
+        self._wait_exit()
         self._exited = True
 
+    def _wait_exit(self) -> None:
+        if self._exit_thread is None:
+            return
+        self._exit_thread.join()
+
+    def stop_async(self):
+        """Return immediately but will stop the server."""
+
+        def stop_f():  # wait for all connections to close
+            while len(self._server.clients) > 0:
+                time.sleep(0.005)
+            self._background_app.stop()
+            self._server.close()
+            self._exited = True
+
+        self._exit_thread = threading.Thread(target=stop_f)
+        self._exit_thread.start()
+
     def stop(self) -> None:
         """Stop the server."""
         print("Stopping the server...")

tests/conftest.py

@@ -27,6 +27,7 @@
 import shutil
 import subprocess
 import sys
+import time
 
 import ansys.tools.path as atp
 import grpc
@@ -305,15 +306,6 @@ def connect_rpc_embedded_server(port: int):
     return client
 
 
-def stop_embeddedd_server():
-    # TODO: not use a terminate.
-    global embedded_server
-    if embedded_server is not None:
-        embedded_server.terminate()
-        embedded_server.wait()
-        embedded_server = None
-
-
 @pytest.fixture(scope="session")
 def mechanical(pytestconfig, rootdir):
     print("current working directory: ", os.getcwd())
@@ -338,9 +330,19 @@ def mechanical(pytestconfig, rootdir):
     print(mechanical)
     yield mechanical
     if is_embedded_server:
-        print("stopping embedded server")
+        print("\n Stopping embedded server")
+        global embedded_server
         mechanical.exit()
-        stop_embeddedd_server()
+        start_time = time.time()
+        while embedded_server.poll() is None:
+            if time.time() - start_time > 10:
+                try:
+                    embedded_server.terminate()
+                    embedded_server.wait()
+                except subprocess.TimeoutExpired:
+                    embedded_server.kill()
+                break
+            time.sleep(0.5)
     else:
         assert "Ansys Mechanical" in str(mechanical)