Rclcpp: Add the ability to have asynchronous service callbacks

[clalancette]

Feature request

Add in an API to do asynchronous service callbacks for service servers.

Feature description

It is currently possible in ROS2 to have a service callback in a thread by using separate callback groups and using a multi-threaded executor. However, it would probably be nice to allow for an API where the user can specify that their service is asynchronous. During the initial call to the service, the service would kick off a thread (or whatever else is needed to do the work), then return some kind of promise. When the service eventually completes, it would signal the client that it has completed the promise. I'm not entirely sure what the API for this would look like currently, but @wjwwood may have some ideas here.

[mkhansenbot]

What is the status of this feature request. I'm looking at how to do this for implementing Actions, and don't want to duplicate.

[wjwwood]

What is the status of this feature request.

Nothing has been done about this feature request as far as I'm aware.

Basically we just need a Service server callback that doesn't require that you return the response from the callback and instead allows you to send the response when ever and from where ever you want.

[mkhansenbot]

I'm not sure where to ask this, here or on Discourse; would it be possible to add another callback group type that is "concurrent non-reentrant"? Right now there are only "MutuallyExclusive" and "Reentrant" types. It would be nice to be able to have multiple callbacks in the same group able to execute concurrently, but not pre-empt themselves (non-reentrant).

[wjwwood]

"concurrent non-reentrant"

Perhaps the name should be changed, but I don't see a need for anything other than the two callback groups we have at the moment.

If you consider two timers, A and B, then you can make a sort of table:

| | Mutually Exclusive | Reentrant |
|------------------------|-------------------------------------------------------------------------|--------------------------------------------------------------------|
| A and B in same CG | A and B will not be called at the same time as each other or themselves | A and B may be called at the same time as each other or themselves |
| A and B in separate CG | A and B will be called concurrently with each other but not themselves | A and B may be called at the same time as each other or themselves |

The only case not served by the current design that I'm aware of is "A and B may be called at the same time as themselves, but not each other", which was deemed uncommon enough that we didn't need to support it. We'd either have to allow nesting callback groups (explicitly decided not to do that originally) or make some other change if we wanted to support that, but I don't think we need to do so.

So I don't think there is a distinction about re-entrant versus concurrent in this case. My perspective is that in general they can be different, but in this case the implication is "they should not share state". This might get confused with reentrancy within the same thread (like with a signal interrupt), but that's not the case here. This callback group might could be renamed ConcurrentCallbackGroup, but the implication I was trying to convey originally was that ideally they would also not synchronize with one another to avoid thread starvation in the executor. I have a slide about this in one of my ROSCon talks:

https://roscon.ros.org/2014/wp-content/uploads/2014/07/ROSCON-2014-Why-you-want-to-use-ROS-2.pdf (starting on page 17)

---

Long story short, I'd recommend using the ReentrantCallbackGroup and synchronizing with itself to maintain thread-safety as little as possible (e.g. lock a data structure access rather than the whole function).

[wjwwood]

Also, I'll point out that this feature request is meant to work around the need for a ReentrantCallbackGroup or even a multi-threaded executor, by allowing the actual callback to be quick and dispatching the long running task to user created threads, finally making and sending the response at a later time and place.

[mkhansenbot]

@wjwwood - thanks for the long answer, I understand it now and am able to use the Reentrant method.

[dirk-thomas]

@stonier FYI

[itfanr]

Has the client callback been implemented?

Could somebody show me a example?

[wjwwood]

As far as I know it has not.

[rpaaron]

If you don't mind duplicating a few layers of existing code:

template<typename ServiceT>
class AsyncService : public rclcpp::Service<ServiceT>
{
  rclcpp::AnyServiceCallback<ServiceT> any_callback_;

  void handle_request(std::shared_ptr<rmw_request_id_t> request_header,
                      std::shared_ptr<void> request) override
  {
    auto typed_request = std::static_pointer_cast<typename ServiceT::Request>(request);
    auto response = std::make_shared<typename ServiceT::Response>();
    // perhaps the callback could return a promise and it go on a queue somewhere?
    any_callback_.dispatch(request_header, typed_request, response);

    std::thread t([=](){
      /* ** takes a long time to respond ** */
      using namespace std::chrono_literals;
      std::this_thread::sleep_for(3s);

      // is this even safe to call this from another thread?
      this->send_response(*request_header, *response);
    });

    t.detach();
  }
public:
  AsyncService(std::shared_ptr<rcl_node_t> node_handle, const std::string & service_name, 
           rclcpp::AnyServiceCallback<ServiceT> any_callback, rcl_service_options_t& service_options)
      : rclcpp::Service<ServiceT>::Service(node_handle, service_name, any_callback, service_options),
    any_callback_(any_callback)
  {}
};

class Node : public rclcpp::Node
{
  using rclcpp::Node::Node;

public:
  template<typename ServiceT, typename CallbackT>
  typename rclcpp::Service<ServiceT>::SharedPtr create_async_service(const std::string& service_name,
                                                      CallbackT&& callback)
  {
    auto qos_profile = rmw_qos_profile_services_default;
    auto group = nullptr;
    auto node_base = get_node_base_interface();
    auto node_services = get_node_services_interface();

    rclcpp::AnyServiceCallback<ServiceT> any_service_callback;
    any_service_callback.set(callback);
    rcl_service_options_t service_options = rcl_service_get_default_options();
    service_options.qos = qos_profile;

    auto serv = std::make_shared<AsyncService<ServiceT>>(
        node_base->get_shared_rcl_node_handle(),
    service_name,
    any_service_callback,
    service_options);

    auto serv_base_ptr = std::dynamic_pointer_cast<rclcpp::ServiceBase>(serv);
    node_services->add_service(serv_base_ptr, group);
    return serv;
  }
};

is it safe to call send_request from some arbitrary thread?
Is there an existing mechanism for the main loop to wait on a promise? or other async structure?

[wjwwood]

is it safe to call send_request from some arbitrary thread?

I think so yes.

Is there an existing mechanism for the main loop to wait on a promise? or other async structure?

You mean like rclcpp::spin_until_future_complete? Maybe I'm misunderstanding the question though.

[rpaaron]

You mean like rclcpp::spin_until_future_complete? Maybe I'm misunderstanding the question though.

I thought it may be necessary to return a future<Response> so that ROS could wait on it and send the response when it becomes available.
But if the user is free to call send_request, this would be simpler. But reasonable to burden the user with this responsibility?

[wjwwood]

I'm sorry, I don't exactly understand what you're asking. What might need to return a future? Do you mean send_response when you're asking if the user is free to call it?

[rpaaron]

I'm sorry, I don't exactly understand what you're asking. What might need to return a future? Do you mean send_response when you're asking if the user is free to call it?

yes, sorry send_response.

[rpaaron]

I'm sorry, I don't exactly understand what you're asking. What might need to return a future? Do you mean send_response when you're asking if the user is free to call it?

What would be the ideal API for this? I have a couple of ideas.

First idea where ROS sends the reply internally, as it does for synchronous callbacks

auto service = node->create_service<AddTwoInts>("add_two_ints", [&](
      const std::shared_ptr<rmw_request_id_t> request_header,
      const std::shared_ptr<AddTwoInts::Request> request,
      const std::shared_ptr<AddTwoInts::Response> response,
) {
  // act of returning a `std::future<AddTwoInts::Response>` means the ROS infrastructure
  // will take care of calling send_response internally.  Assumes there is some mechanism 
  // to do that already, or it needs to be written?
  return std::async(std::launch::async, [=]{ 
    response->sum = 42;
    return response; 
  });
});

Second idea, where its the users responsibility to finish the job

auto service_holder = node->create_service<AddTwoInts>("add_two_ints", [&](
      const std::shared_ptr<rmw_request_id_t> request_header,
      const std::shared_ptr<AddTwoInts::Request> request,
      const std::shared_ptr<AddTwoInts::Response> response,
      const std::shared_ptr<Service<AddTwoInts>> service // indication of async callback by requiring service in sig
) {
  std::async(std::launch::async, [=]{ 
    response->sum = 42;
    service->send_response(request_header, response);    // user is responsible for calling `send_response`
  });                                                    // no extra infrastructure required
});

[rpaaron]

perhaps the least invasive approach is to have a different method name? eg create_async_service

// but can't use `auto` if we want to use `service` in the lambda
// and can't use `bind` to a global method as it won't have access to `service`
Service<AddTwoInts> service = node->create_async_service<AddTwoInts>("add_two_ints", [&](
      const std::shared_ptr<rmw_request_id_t> request_header,
      const std::shared_ptr<AddTwoInts::Request> request,
      const std::shared_ptr<AddTwoInts::Response> response,
) {
  auto callback =  [=](result) {
    response->sum = result;
    service->send_response(request_header, response);
  };
  call_some_external_async_add_service(request.a, request.b, callback);
});