高精度线程同步

High precision thread synchronization

本文关键字：同步 线程 高精度      更新时间：2023-10-16

我有几个运行网络数据交换的类的对象。我需要尽可能准确地在线程中启动网络操作。这是以微秒为单位。工作开始时，线程已创建并正在运行。对于同步，我使用静态原子变量。在主线程中创建的对象，然后在其他线程中创建的对象启动这些对象的函数 on_start((。

class NetSender {
public: 
  static std::atomic_flag start;
  static std::atomic<uint8_t> net_sender_count;
  void on_start(){
    //... do some work
    net_sender_count++;        
    //waiting start of other 
    while(start.test_and_set())
        ;
    m_start_message = std::chrono::system_clock::now();
    //do very important job here
  }
}

和 main(( 函数中的某个地方

std::vector<NetSender*> senders[how_much_we_need];
//create senders items
std::vector<std::thread> senderthreads[how_much_we_need]; 
// for senders
senderthreads.at(i) = new std::thread(
                &NetSender::on_start,senders.at(i));
while(senders[1]->net_sender_count < how_much_we_need)
     ; //whait starting of all threads
senders[1]->start.clear();    //start job

(也许我的例子中有一些错别字，但我希望这个想法非常明显(

之后，我比较m_start_message值，差异从 10 微秒到 100 毫秒不等。有没有办法将这种差异减少到微秒？或者可能存在一些逻辑错误。

您需要一个条件变量和一个互斥组合。

如下所示：

class NetSender
{
    static std::condition_variable _cvThreads; // threads signal this cv when ready
    static std::condition_variable _cvMain;    // main signals this cv after setting start condition
    static std::mutex _mutex;
    static bool _isStarted;
    static int net_sender_count;
    static int how_much_we_need;
    void on_start()
    {
        { // enter lock
            std::lock_guard<std::mutex> lck(_mutex);
            net_sender_count++;
        } // exit lock
        _cvThreads.notify_all();  // notify main thread that this worker thread is ready
        // wait for main thread to signal start
        { // enter lock
            std::lock_guard<std::mutex> lck(_mutex);
            while (_isStarted == false)
            {
                _cvMain.wait(lck); // atomically release mutex, wait for notify, then lock and continue
            }
        } // exit lock
        m_start_message = std::chrono::system_clock::now();
        //do very important job here
    }

然后在您的主线程中

NetSender::_isStarted = false;
senderthreads.at(i) = new std::thread(&NetSender::on_start,senders.at(i));
// wait for all threads to get ready
{
    std::lock_guard<std::mutex> lck(NetSender::_mutex);
    while (NetSender::net_sender_count < NetSender::how_many_we_need)
    {
         NetSender::_cvThreads.wait();
    }
    NetSender::_isStarted = true;
}
NetSender::_cvMain.notify_all(); // signal all threads to start

一些提示来使其工作：

1( 如果您要多次执行，则可能需要使用线程池。

2( 您可能不想使用等待/通知，因为这会在取消调度/调度线程时引入延迟。如果等待时间足够短，最好只是旋转。

3(你正在寻找的是线屏障。有许多已知的算法，您可能想利用它们。如果这是一次性执行，则倒计时闩锁可能是有利的。否则，可以重复使用诸如反感障碍之类的障碍。如有必要，请使用树形屏障来最大程度地减少内核之间的一致性流量。

一个简单的实现可能看起来像这样(伪代码，对不起，我已经有一段时间没有写C++了(：

void main() {
  // Create a barrier sufficient for syncing worker threads + main thread
  MyBarrier * b = new MyBarrier(num_threads + 1);
  std::thread threads[num_threads];
  for (int i = 0; i < num_threads; i++) {
    threads[i] = thread(onStart, &senders[i], b);
  }
  someWork();
  barrier->wait();
  ...
}
void onStart(NetSender *sender, MyBarrier * b) {
  doSomeWork();
  b->wait();
  doSomeMoreWork(); 
}
class MyBarrier {
private:
  std::atomic<int> count;
public:
  MyBarrier(int countdown) {
    std::atomic_init(&count, countdown);
  }
  void wait() {
    count.fetch_sub(1);
    while (count.get() > 0) {
      // spin
    }
  }
}
相关文章：