避免繁忙的等待和实时和非实时线程之间的模式切换

Question

我有以下问题:我们有一个用ros_control实现的控制器，它运行在一个实时的Xenomai linux补丁系统上。通过迭代调用更新函数来执行控制循环。我需要传达控制器的一些内部状态，为此我使用了麻省理工学院开发的LCM。不管LCM的内部行为如何，发布方法破坏了实时性，因此我在C 11中实现了一个在独立线程上运行的发布循环。但是如果我不把第二个线程和控制器同步，这个循环就会以无限的频率发布。因此我也使用了条件变量。

以下是控制器的一个示例:

MyClass mc;

// This is called just once
void init(){
    mc.init();
}

// Control loop function (e.g., called every 5 ms in RT)
void update(const ros::Time& time, const ros::Duration& period) {
    double value = time.toSec();
    mc.setValue(value);
}

对于尝试发布的类：

double myvalue;
std::mutex mutex;
std::condition_variable cond;
bool go = true;

void MyClass::init(){
    std::thread thread(&MyClass::body, this);
}


void MyClass::setValue(double value){
    myvalue = value;
    {
    std::lock_guard<std::mutex> lk(mutex);
    go = true;
    }
    cond.notify_one();
} 

void MyClass::body() {
    while(true) {
        std::unique_lock<std::mutex>lk(mutex);
        cond.wait(lk, [this] {return go;});    
        publish(myvalue); // the dangerous call
        go = false;
        lk.unlock();
    }
 }

此代码产生模式切换（即实时中断）。可能是因为条件变量的锁定，我用它来同步辅助线程与主控制器，并与线程争用。如果我这样做：

void MyClass::body() {
    while(true) {
        if(go){
        publish(myvalue);
        go = false;            
        }

    }
 }
 void MyClass::setValue(double value){
    myvalue = value;
    go = true;        
} 

我不会产生模式切换，但这也不安全，最重要的是，我会忙于等待第二个线程。

有没有一种方法可以在主线程和辅助线程之间实现非阻塞同步（即只有在调用setValue时才让body执行某些操作），这也是非繁忙的等待？

Answer 1

这并不完美，但它应该减少您的繁忙等待频率，只是偶尔失去响应能力。

这个想法是在通过原子传递消息时使用裸条件变量唤醒。

template<class T>
struct non_blocking_poke {
  std::atomic<T> message;
  std::atomic<bool> active;
  std::mutex m;
  std::condition_variable v;

  void poke(T t) {
    message = t;
    active = true;
    v.notify_one();
  }
  template<class Rep, class Period>
  T wait_for_poke(const std::chrono::duration<Rep, Period>& busy_time) {
    std::unique_lock<std::mutex> l(m);
    while( !v.wait_for(l, busy_time, [&]{ return active; } ))
    {}
    active = false;
    return message;
  }
};

等待线程每隔busy_time唤醒一次，以查看它是否错过了一条消息。但是，它通常会比这更快地收到消息（存在一个争用条件，即它错过了一条消息）。此外，可以发送多条消息，而无需缓解程序获取它们。但是，如果发送了一条消息，则在大约 1 秒内，接收方将收到该消息或稍后的消息。

non_blocking_poke<double> poker;

// in realtime thread:
poker.poke(3.14);

// in non-realtime thread:
while(true) {
  using namespace std::literals::chrono_literals;
  double d = poker.wait_for_poke( 1s );
  std::cout << d << '\n';
}

在工业质量解决方案中，您还需要中止标志或消息来停止循环。

Answer 2

使用无锁数据结构。

在你这里的情况下，你甚至不需要一个数据结构，只需使用一个原子就可以了。无需锁。您可以考虑使用信号量而不是条件变量来避免现在未使用的锁。如果你需要一个信号量来避免使用锁，你最终会使用你的基本操作系统信号量，而不是C 11，因为C 11甚至没有它们。