feat: adding python side retry mechanism (#3354)

* feat: adding python retry/reconnect mechanism

* feat: connecting only one time per retry

* fix: avoid running commands when the instance has exited and using exited function more broadly.

* feat: check if MAPDL has died properly during processes kill

* fix: looping

* refactor: error message

* chore: adding changelog file 3354.miscellaneous.md

* chore: adding changelog file 3354.dependencies.md

* chore: adding changelog file 3354.miscellaneous.md

* fix: test

* fix: wrong name in variable.

---------

Co-authored-by: pyansys-ci-bot <[email protected]>

Author: germa89

doc/changelog.d/3354.miscellaneous.md

@@ -0,0 +1 @@
+feat: adding python side retry mechanism

src/ansys/mapdl/core/errors.py

@@ -24,6 +24,7 @@
 from functools import wraps
 import signal
 import threading
+from time import sleep
 from typing import Callable, Optional
 
 import grpc
@@ -306,26 +307,70 @@ def wrapper(*args, **kwargs):
                 old_handler = signal.signal(signal.SIGINT, handler)
 
         # Capture gRPC exceptions
-        try:
-            out = func(*args, **kwargs)
-        except grpc.RpcError as error:
-            # Custom errors
-            if error.code() == grpc.StatusCode.RESOURCE_EXHAUSTED:
-                if "Received message larger than max" in error.details():
-                    try:
-                        lim_ = int(error.details().split("(")[1].split("vs")[0])
-                    except IndexError:
-                        lim_ = int(512 * 1024**2)
-
-                    raise MapdlgRPCError(
-                        f"RESOURCE_EXHAUSTED: {error.details()}. "
-                        "You can try to increase the gRPC message length size using 'PYMAPDL_MAX_MESSAGE_LENGTH'"
-                        " environment variable. For instance:\n\n"
-                        f"$ export PYMAPDL_MAX_MESSAGE_LENGTH={lim_}"
+        n_attempts = 3
+        initial_backoff = 0.05
+        multiplier_backoff = 3
+
+        i_attemps = 0
+
+        while True:
+            try:
+                out = func(*args, **kwargs)
+
+                # Exit while-loop if success
+                break
+
+            except grpc.RpcError as error:
+
+                mapdl = retrieve_mapdl_from_args(args)
+
+                i_attemps += 1
+                if i_attemps <= n_attempts:
+
+                    wait = (
+                        initial_backoff * multiplier_backoff**i_attemps
+                    )  # Exponential backoff
+                    sleep(wait)
+
+                    # reconnect
+                    mapdl._log.debug(
+                        f"Re-connection attempt {i_attemps} after waiting {wait:0.3f} seconds"
+                    )
+
+                    connected = mapdl._connect(timeout=wait)
+
+                    # Retry again
+                    continue
+
+                # Custom errors
+                reason = ""
+                suggestion = ""
+
+                if error.code() == grpc.StatusCode.RESOURCE_EXHAUSTED:
+                    if "Received message larger than max" in error.details():
+                        try:
+                            lim_ = int(error.details().split("(")[1].split("vs")[0])
+                        except IndexError:
+                            lim_ = int(512 * 1024**2)
+
+                        raise MapdlgRPCError(
+                            f"RESOURCE_EXHAUSTED: {error.details()}. "
+                            "You can try to increase the gRPC message length size using 'PYMAPDL_MAX_MESSAGE_LENGTH'"
+                            " environment variable. For instance:\n\n"
+                            f"$ export PYMAPDL_MAX_MESSAGE_LENGTH={lim_}"
+                        )
+
+                if error.code() == grpc.StatusCode.UNAVAILABLE:
+                    # Very likely the MAPDL server has died.
+                    suggestion = (
+                        "  MAPDL *might* have died because it executed a not-allowed command or ran out of memory.\n"
+                        "  Check the MAPDL command output for more details.\n"
+                        "  Open an issue on GitHub if you need assistance: "
+                        "https://github.com/ansys/pymapdl/issues"
                     )
 
-            # Generic error
-            handle_generic_grpc_error(error, func, args, kwargs)
+                # Generic error
+                handle_generic_grpc_error(error, func, args, kwargs, reason, suggestion)
 
         # No exceptions
         if threading.current_thread().__class__.__name__ == "_MainThread":
@@ -344,15 +389,26 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-def handle_generic_grpc_error(error, func, args, kwargs):
-    """Handle non-custom gRPC errors"""
-
+def retrieve_mapdl_from_args(args):
     # can't use isinstance here due to circular imports
     try:
         class_name = args[0].__class__.__name__
     except (IndexError, AttributeError):
         class_name = ""
 
+    if class_name == "MapdlGrpc":
+        mapdl = args[0]
+    elif hasattr(args[0], "_mapdl"):
+        mapdl = args[0]._mapdl
+
+    return mapdl
+
+
+def handle_generic_grpc_error(error, func, args, kwargs, reason="", suggestion=""):
+    """Handle non-custom gRPC errors"""
+
+    mapdl = retrieve_mapdl_from_args(args)
+
     # trying to get "cmd" argument:
     cmd = args[1] if len(args) >= 2 else ""
     cmd = kwargs.get("cmd", cmd)
@@ -364,28 +420,30 @@ def handle_generic_grpc_error(error, func, args, kwargs):
     else:
         msg_ = f"calling:{caller}\nwith the following arguments:\n  args: {args}\n  kwargs: {kwargs}"
 
-    if class_name == "MapdlGrpc":
-        mapdl = args[0]
-    elif hasattr(args[0], "_mapdl"):
-        mapdl = args[0]._mapdl
+    if reason:
+        reason = f"Possible reason:\n{reason}\n"
+
+    if suggestion:
+        suggestion = f"Suggestions:\n{suggestion}\n"
 
     msg = (
         f"Error:\nMAPDL server connection terminated unexpectedly while {msg_}\n"
+        f"{reason}"
+        f"{suggestion}"
         "Error:\n"
         f"  {error.details()}\n"
         f"Full error:\n{error}"
     )
 
-    # MAPDL gRPC is unavailable.
-    if error.code() == grpc.StatusCode.UNAVAILABLE:
-        raise MapdlExitedError(msg)
-
     # Generic error
     # Test if MAPDL is alive or not.
     if mapdl.is_alive:
         raise MapdlRuntimeError(msg)
 
     else:
+        # Making sure we do not keep executing gRPC calls.
+        mapdl._exited = True
+
         # Must close unfinished processes
         mapdl._close_process()
         raise MapdlExitedError(msg)

src/ansys/mapdl/core/mapdl_core.py

@@ -56,6 +56,7 @@
 from ansys.mapdl.core.errors import (
     ComponentNoData,
     MapdlCommandIgnoredError,
+    MapdlExitedError,
     MapdlFileNotFoundError,
     MapdlInvalidRoutineError,
     MapdlRuntimeError,
@@ -434,7 +435,7 @@ def components(self) -> "ComponentManager":
 
         >>> mapdl.solution.converged
         """
-        if self._exited:  # pragma: no cover
+        if self.exited:  # pragma: no cover
             raise MapdlRuntimeError("MAPDL exited.")
         return self._componentmanager
 
@@ -844,7 +845,7 @@ def post_processing(self) -> "PostProcessing":
         array([1.07512979e-04, 8.59137773e-05, 5.70690047e-05, ...,
                5.70333124e-05, 8.58600402e-05, 1.07445726e-04])
         """
-        if self._exited:
+        if self.exited:
             raise MapdlRuntimeError(
                 "MAPDL exited.\n\nCan only postprocess a live " "MAPDL instance."
             )
@@ -963,7 +964,7 @@ def solution(self) -> "Solution":
 
         >>> mapdl.solution.converged
         """
-        if self._exited:
+        if self.exited:
             raise MapdlRuntimeError("MAPDL exited.")
         return self._solution
 
@@ -2110,6 +2111,11 @@ def run(
         >>> mapdl.prep7()
 
         """
+        if self.exited:
+            raise MapdlExitedError(
+                f"The MAPDL instance has been exited before running the command: {command}"
+            )
+
         # check if multiline
         if "\n" in command or "\r" in command:
             raise ValueError("Use ``input_strings`` for multi-line commands")

src/ansys/mapdl/core/mapdl_grpc.py

@@ -534,9 +534,9 @@ def _multi_connect(self, n_attempts=5, timeout=15):
         attempt_timeout = int(timeout / n_attempts)
 
         max_time = time.time() + timeout
-        i = 0
+        i = 1
         while time.time() < max_time and i <= n_attempts:
-            self._log.debug("Connection attempt %d", i + 1)
+            self._log.debug("Connection attempt %d", i)
             connected = self._connect(timeout=attempt_timeout)
             i += 1
             if connected:
@@ -564,7 +564,7 @@ def _multi_connect(self, n_attempts=5, timeout=15):
                     else ""
                 )
                 raise MapdlConnectionError(
-                    msg + f"The MAPDL process has died{pid_msg}."
+                    msg + f" The MAPDL process has died{pid_msg}."
                 )
 
         self._exited = False
@@ -1194,6 +1194,11 @@ def _close_process(self, timeout=2):  # pragma: no cover
             # Killing child processes
             self._kill_child_processes(timeout=timeout)
 
+        if self.is_alive:
+            raise MapdlRuntimeError("MAPDL could not be exited.")
+        else:
+            self._exited = True
+
     def _cache_pids(self):
         """Store the process IDs used when launching MAPDL.
 

tests/test_grpc.py

@@ -36,6 +36,7 @@
     MapdlExitedError,
     MapdlgRPCError,
     MapdlRuntimeError,
+    protect_grpc,
 )
 from ansys.mapdl.core.mapdl_grpc import MAX_MESSAGE_LENGTH, MapdlGrpc
 from ansys.mapdl.core.misc import random_string
@@ -48,6 +49,19 @@
 pytestmark = requires("grpc")
 
 
+class UnavailableError(grpc.RpcError):
+    def __init__(self, message="Service is temporarily unavailable."):
+        self._message = message
+        self._code = grpc.StatusCode.UNAVAILABLE
+        super().__init__(message)
+
+    def code(self):
+        return self._code
+
+    def details(self):
+        return self._message
+
+
 def write_tmp_in_mapdl_instance(mapdl, filename, ext="txt"):
     """Write a temporary file from MAPDL."""
     with mapdl.non_interactive:
@@ -627,26 +641,45 @@ def test_generic_grpc_exception(monkeypatch, grpc_channel):
     mapdl = MapdlGrpc(channel=grpc_channel)
     assert mapdl.is_alive
 
-    class UnavailableError(grpc.RpcError):
-        def __init__(self, message="Service is temporarily unavailable."):
-            self._message = message
-            self._code = grpc.StatusCode.UNAVAILABLE
-            super().__init__(message)
+    @protect_grpc
+    def _raise_error_code(*args, **kwargs):
+        raise UnavailableError()
 
-        def code(self):
-            return self._code
+    # Monkey patch to raise the same issue.
+    monkeypatch.setattr(mapdl, "prep7", _raise_error_code)
 
-        def details(self):
-            return self._message
+    with pytest.raises(
+        MapdlRuntimeError, match="MAPDL server connection terminated unexpectedly while"
+    ):
+        # passing mapdl to simulate the function `_raise_error_code` to be a method.
+        mapdl.prep7(mapdl)
+
+    assert mapdl.is_alive
 
-    def _raise_error_code(args, **kwargs):
+
+def test_generic_grpc_exception_exited(monkeypatch, grpc_channel):
+    mapdl = MapdlGrpc(channel=grpc_channel)
+    assert mapdl.is_alive
+
+    @protect_grpc
+    def _raise_error_code(*args, **kwargs):
         raise UnavailableError()
 
-    monkeypatch.setattr(mapdl._stub, "SendCommand", _raise_error_code)
+    def _null_close_process():
+        return None
+
+    # faking exiting MAPDL
+    mapdl._exited = True
+
+    # Monkey patch to raise the same issue.
+    monkeypatch.setattr(mapdl, "prep7", _raise_error_code)
+
+    # monkey patch `_close_process` so MAPDL does not exit when
+    monkeypatch.setattr(mapdl, "_close_process", _null_close_process)
 
     with pytest.raises(
         MapdlExitedError, match="MAPDL server connection terminated unexpectedly while"
     ):
-        mapdl.prep7()
+        mapdl.prep7(mapdl)
 
-    assert mapdl.is_alive
+    mapdl._exited = False  # Restoring