高精度线程同步

Question

我有几个运行网络数据交换的类的对象。我需要在线程中启动网络操作，就像possible.This以微秒为单位一样精确。到工作开始时，线程已经创建并运行。对于同步，我使用静态原子变量。在主线程中创建的对象，然后在其他线程中启动这些对象的函数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毫秒。有没有办法将这种差异减小到微秒级？或者可能有一些逻辑上的错误。

Answer 1

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

类似于以下内容：

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

Answer 2

这里有几个让它工作的提示：

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

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

3)你要找的是一个线程屏障。有许多已知的算法，您可能想要利用它们。如果这是一次性执行，倒计时锁存器可能更好。否则，诸如意义颠倒障碍之类的障碍可以被重用。如有必要，使用树屏障将核心之间的一致性流量降至最低。

一个简单的实现可能如下所示(psuedocode，对不起，自从我编写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
    }
  }
}