ROS AsyncSpinner 的多线程行为

Question

我试图了解 ROS 的 AsyncSpinner 是如何工作的，因为我可能有一些误解。你可以找到类似的问题here。

正如here 所见，其定义中提到：

异步微调器： 产生几个线程（可配置） 将并行执行回调而不阻塞线程 叫它。 start/stop 方法允许控制回调的时间 开始处理，何时停止。

在官方文档here 中，AsyncSpinning 也被标注为一种多线程 Spinning。

也就是说，我有一个非常简单的示例，发布者和订阅者使用 AsyncSpinner 来测试多线程行为。

#include "ros/ros.h"
#include "std_msgs/String.h"

int main(int argc, char **argv)
{
  ros::init(argc, argv, "publisher");
  ros::NodeHandle nh;

  ros::Publisher chatter_pub = nh.advertise<std_msgs::String>("chatter", 1000);

  ros::Rate loop_rate(10);
  while (ros::ok())
  {
    std_msgs::String msg;
    msg.data = "hello world";

    chatter_pub.publish(msg);

    ros::spinOnce();

    loop_rate.sleep();
  }

  return 0;
}

以及定义和使用微调器的订阅者：

#include "ros/ros.h"
#include "std_msgs/String.h"
#include <boost/thread.hpp>

int count = 0;

void chatterCallback(const std_msgs::String::ConstPtr& msg)
{
  count++;
  ROS_INFO("Subscriber %i callback: I heard %s", count, msg->data.c_str());
  sleep(1);
}

int main(int argc, char **argv)
{
  ros::init(argc, argv, "subscriber");
  ros::NodeHandle nh;

  ros::Subscriber sub = nh.subscribe("chatter", 1000, chatterCallback);

  ros::AsyncSpinner spinner(boost::thread::hardware_concurrency());
  ros::Rate r(10);

  spinner.start();
  ros::waitForShutdown();

  return 0;
}

当我运行这两个程序时，我得到以下输出：

[ INFO] [1517215527.481856914]: Subscriber 1 callback: I heard hello world
[ INFO] [1517215528.482005146]: Subscriber 2 callback: I heard hello world
[ INFO] [1517215529.482204798]: Subscriber 3 callback: I heard hello world

如您所见，回调每秒运行一次，并且没有并行调用其他回调。我知道全局回调队列正在执行，因为如果我停止发布者，订阅者将继续从队列中弹出累积的消息。

我知道我不应该阻止回调，但在上面的定义中指出，这不会停止调用它的线程，我猜其他人也不会停止由微调器创建的线程。我是否只是因为我阻止了回调而阻止了下一个回调？我有什么误解吗？我有点困惑，无法证明回调是并行运行的。也许你还有另一个例子？

Answer 1

简答：

ROS 回调默认是线程安全的。这意味着注册的回调只能由一个线程处理，并发调用被禁用。第二个线程不能同时访问同一个回调。

如果您注册第二个回调，您将看到微调器按预期工作，并且多个线程同时调用您的回调。

ros::Subscriber sub1 = nh.subscribe("chatter", 1000, chatterCallback);
ros::Subscriber sub2 = nh.subscribe("chatter", 1000, chatterCallback);

扩展答案：

异步微调器会以允许的速率尽可能快地尝试call 回调队列中的可用回调。如果回调已经在处理中（由另一个线程），CallResult 是TryAgain。这意味着稍后将开始新的尝试。

implementation of this lock 使用变量allow_concurrent_callbacks_，这意味着此行为是可选的。

解决方案：

可以通过设置正确的SubscribeOptions.allow_concurrent_callbacks 来允许并发调用，默认为false。因此您需要定义自己的SubscribeOptions。这是您需要订阅并允许并发回调调用的代码：

ros::SubscribeOptions ops;
ops.template init<std_msgs::String>("chatter", 1000, chatterCallback);
ops.transport_hints = ros::TransportHints();
ops.allow_concurrent_callbacks = true;
ros::Subscriber sub = nh.subscribe(ops);