Services feature parity

[mikaelarguedas]

Feature request

Feature description

Meta ticket to track improvements we want to make to services and service-related features. This is not an exhaustive list, feel free to complete it

• create a "graph_listener" equivalent to use for wait_for_service (related to Asyncronous wait_for_service #58)
• refactor client to store pending requests
• refactor clients to return a future for each request sent and provide an equivalent of rclcpp::spin_until_future_complete() at the executor level

connects to #170

[sloretz]

I see Executor::spin_until_future_complete(), but not rclcpp::spin_until_future_complete(). How is it meant to be used?

Using it in the main thread outside of an executor looks fine.

future = client.call(req)
executor.spin_until_future_complete(future)
if future.get() is not None:
    print('Doing something with result %r', future.get())

Used in a callback done by a single threaded executor would be ok, except from the perspective of the main thread a spin_once() call might handle multiple callbacks. From a multi threaded executor it would cause rcl_wait() to be called on the same wait_set in multiple threads.

# main thread
while rclpy.ok()
    # OK, though spin_once may handle more than 1 callback since the callback can spin
    singleThreadedExecutor.spin_once()
    # Not OK, this will dispatch the callback and call rcl_wait again,
    # but the callback will also call spin which will call rcl_wait on the
    # same wait set in another thread. rcl_wait() warns it is not thread safe.
    # multiThreadedExecutor.spin_once()

# ...

# Subscriber callback
    def listener_cb(self, response):
        future = self.client.call(req)
       # Executor has to be given to this object somehow
        self.executor.spin_until_future_complete(future)
        if future.get() is not None:
            print('Doing something with result %r', future.get())

[sloretz]

Thoughts/notes regarding Future objects:

• Every Future needs a Task to set the result of the Future
  • These need to be added to the Executor so it can schedule them
    • This seems to be the reason but concurrent.futures and asyncio recommend using the executor to create futures
  • client.call(req) could returns a future and keep a list of tasks. Executor would pull those tasks off the client object
• Need to either subclass or create a custom Future class to support both multithreading and asynchronous execution
  • concurrent.futures.Future is thread safe but not awaitable
  • asyncio.Future is awaitable but not thread safe
• ...

[sloretz]

Connected to #170 because that implements almost everything in the description. #58 (asyncronous wait_for_service) will carry on on it's own.