Merge pull request #1574 from jluebbe/improve-grpc-robustness

improve gRPC robustness

Author: Bastian-Krause

labgrid/remote/coordinator.py

@@ -169,13 +169,22 @@ def __init__(self, request) -> None:
         self.request = request
         self.response = None
         self.completed = asyncio.Event()
+        self.expired = False
 
     def complete(self, response) -> None:
         self.response = response
         self.completed.set()
+        if self.expired:
+            logging.warning(
+                "exporter command already expired for request %s -> response %s", self.request, self.response
+            )
 
     async def wait(self):
-        await asyncio.wait_for(self.completed.wait(), 10)
+        try:
+            await asyncio.wait_for(self.completed.wait(), 10)
+        finally:
+            if self.response is None:
+                self.expired = True
 
 
 class ExporterError(Exception):
@@ -592,7 +601,9 @@ async def _acquire_resource(self, place, resource):
         self.get_exporter_by_name(resource.path[0]).queue.put_nowait(cmd)
         await cmd.wait()
         if not cmd.response.success:
-            raise ExporterError("failed to acquire {resource}")
+            raise ExporterError("failed to acquire {resource} ({cmd.response.reason})")
+        if resource.acquired != place.name:
+            logging.warning("resource %s not acquired by this place after acquire request", resource)
 
     async def _acquire_resources(self, place, resources):
         assert self.lock.locked()
@@ -646,7 +657,9 @@ async def _release_resources(self, place, resources, callback=True):
                     self.get_exporter_by_name(resource.path[0]).queue.put_nowait(cmd)
                     await cmd.wait()
                     if not cmd.response.success:
-                        raise ExporterError(f"failed to release {resource}")
+                        raise ExporterError(f"failed to release {resource} ({cmd.response.reason})")
+                    if resource.acquired:
+                        logging.warning("resource %s still acquired after release request", resource)
             except (ExporterError, TimeoutError):
                 logging.exception("failed to release %s", resource)
                 # at leaset try to notify the clients

labgrid/remote/exporter.py

@@ -37,6 +37,14 @@ class BrokenResourceError(ExporterError):
     pass
 
 
+class UnknownResourceError(ExporterError):
+    pass
+
+
+class InvalidResourceRequestError(ExporterError):
+    pass
+
+
 def log_subprocess_kernel_stack(logger, child):
     if child.poll() is not None:  # nothing to check if no longer running
         return
@@ -889,8 +897,9 @@ async def message_pump(self):
                                 out_message.set_acquired_request.resource_name,
                             )
                         success = True
-                    except BrokenResourceError as e:
+                    except (BrokenResourceError, InvalidResourceRequestError, UnknownResourceError) as e:
                         reason = e.args[0]
+                        logging.warning("set_acquired_request failed: %s", reason)
                     finally:
                         in_message = labgrid_coordinator_pb2.ExporterInMessage()
                         in_message.response.success = success
@@ -924,8 +933,14 @@ async def message_pump(self):
     async def acquire(self, group_name, resource_name, place_name):
         resource = self.groups.get(group_name, {}).get(resource_name)
         if resource is None:
-            logging.error("acquire request for unknown resource %s/%s by %s", group_name, resource_name, place_name)
-            return
+            raise UnknownResourceError(
+                f"acquire request for unknown resource {group_name}/{resource_name} by {place_name}"
+            )
+
+        if resource.acquired:
+            raise InvalidResourceRequestError(
+                f"Resource {group_name}/{resource_name} is already acquired by {resource.acquired}"
+            )
 
         try:
             resource.acquire(place_name)
@@ -935,8 +950,10 @@ async def acquire(self, group_name, resource_name, place_name):
     async def release(self, group_name, resource_name):
         resource = self.groups.get(group_name, {}).get(resource_name)
         if resource is None:
-            logging.error("release request for unknown resource %s/%s", group_name, resource_name)
-            return
+            raise UnknownResourceError(f"release request for unknown resource {group_name}/{resource_name}")
+
+        if not resource.acquired:
+            raise InvalidResourceRequestError(f"Resource {group_name}/{resource_name} is not acquired")
 
         try:
             resource.release()