Fix multi-threaded race condition in client.call_async (ros2#871)

* fix multi-thread race condition in client.call

(cherry picked from commit 301953d)

Author: jmarsik

rclpy/rclpy/client.py

@@ -64,6 +64,8 @@ def __init__(
         # True when the callback is ready to fire but has not been "taken" by an executor
         self._executor_event = False
 
+        self._lock = threading.Lock()
+
     def call(self, request: SrvTypeRequest) -> SrvTypeResponse:
         """
         Make a service request and wait for the result.
@@ -93,22 +95,6 @@ def unblock(future):
             raise future.exception()
         return future.result()
 
-    def remove_pending_request(self, future: Future) -> None:
-        """
-        Remove a future from the list of pending requests.
-
-        This prevents a future from receiving a response and executing its done callbacks.
-
-        :param future: A future returned from :meth:`call_async`
-        """
-        for seq, req_future in self._pending_requests.items():
-            if future == req_future:
-                try:
-                    del self._pending_requests[seq]
-                except KeyError:
-                    pass
-                break
-
     def call_async(self, request: SrvTypeRequest) -> Future:
         """
         Make a service request and asyncronously get the result.
@@ -122,18 +108,44 @@ def call_async(self, request: SrvTypeRequest) -> Future:
         if not isinstance(request, self.srv_type.Request):
             raise TypeError()
 
-        with self.handle:
-            sequence_number = self.__client.send_request(request)
-        if sequence_number in self._pending_requests:
-            raise RuntimeError('Sequence (%r) conflicts with pending request' % sequence_number)
+        with self._lock:
+            with self.handle:
+                sequence_number = self.__client.send_request(request)
+            if sequence_number in self._pending_requests:
+                raise RuntimeError(f'Sequence ({sequence_number}) conflicts with pending request')
 
-        future = Future()
-        self._pending_requests[sequence_number] = future
+            future = Future()
+            self._pending_requests[sequence_number] = future
 
-        future.add_done_callback(self.remove_pending_request)
+            future.add_done_callback(self.remove_pending_request)
 
         return future
 
+    def get_pending_request(self, sequence_number: int) -> Future:
+        """
+        Get a future from the list of pending requests.
+
+        :param sequence_number: Number identifying the pending request.
+        :return: The future corresponding to the sequence_number.
+        :raises: KeyError if the sequence_number is not in the pending requests.
+        """
+        with self._lock:
+            return self._pending_requests[sequence_number]
+
+    def remove_pending_request(self, future: Future) -> None:
+        """
+        Remove a future from the list of pending requests.
+
+        This prevents a future from receiving a response and executing its done callbacks.
+
+        :param future: A future returned from :meth:`call_async`
+        """
+        with self._lock:
+            for seq, req_future in self._pending_requests.items():
+                if future is req_future:
+                    del self._pending_requests[seq]
+                    break
+
     def service_is_ready(self) -> bool:
         """
         Check if there is a service server ready.

rclpy/rclpy/executors.py

@@ -351,7 +351,7 @@ async def _execute_client(self, client, seq_and_response):
         if header is not None:
             try:
                 sequence = header.request_id.sequence_number
-                future = client._pending_requests[sequence]
+                future = client.get_pending_request(sequence)
             except KeyError:
                 # The request was cancelled
                 pass