具有组独占控制并允许组内并发的 C++14 多线程

Question

您好，我遇到了一个多线程场景的问题：

假设我有 3 个线程在一种函数（命名为 node_a）上运行，3 个线程在另一种函数（命名为 node_b）上运行。每个线程处理不同的输入数据。他们都在某个时间访问共享资源。

现在我要控制这些线程的访问：3个node_a线程被认为是grp_a，3个node_b线程被认为是grp_b。 grp_a和grp_b是互斥访问共享资源的，但是grp_a和grp_b本身可以同时运行。

这是我的方法：

• 我使用了两个共享计数器并使用互斥锁控制它们
• 我将 shared_timed_mutex 用于独占，将 shared 用于并发和独占

但事实证明，我没有通过打印每个线程访问共享资源的时间戳来获得独占。

下面是我的 C++14 代码：

    std::mutex mtx;
    std::shared_timed_mutex mtx_a,mtx_b;
    atomic_int cnt_a{0},cnt_b{0};
    
    //code for node_a 
      mtx_a.lock_shared();//will block if lane holds the lock.+1
      bool a=0;
      while(cnt_a==0)  { //check if node_b threads are not running
        if(mtx_b.try_lock()) {
          std::unique_lock<std::mutex> lck(mtx);
          cnt_a++;//multiple threads change it, need protection!
          a=1;
        }
      }
    
    /*
    do some work here!!!!
    */
    
      mtx_a.unlock_shared();//-1
      {
        std::unique_lock<std::mutex> lck(mtx);
        --cnt_a;
      }
      if(a){
        mtx_a.lock(); //block until all node_a threads are done
        mtx_b.unlock(); //can only be unlocked by current thread
        cnt_a=0;
        mtx_a.unlock();
      }

node_b 的类似代码

有人知道是什么问题吗？谢谢

Answer 1

您的实施存在一些问题。

• 使用定时互斥体是可疑的
• 当你有一个线程需要唤醒以响应某个事件时，condition_variables、futures 或其他一些阻塞同步就应该参与进来。现在显然不是使用自旋锁的时候。
• 必须使用互斥锁来保护原子变量并不是一个好兆头。

我建议您将程序视为状态机。它可以处于以下三种状态之一：

1. 不工作（空闲）
2. 执行 a 组的工作。
3. 执行 b 组的工作。

每当一个线程想要获取资源时，程序当前所处的这些状态中的哪一个决定该线程是被允许继续还是等待状态回到空闲状态。

就同步资源而言，您应该需要：

• 用于跟踪状态机当前状态的枚举
• int 用于了解当前有多少线程将机器“保持”在当前状态
• mutex 用于同步对它们的访问。
• condition_variable 等到我们回到空闲状态。

这应该是您需要的全部。仅此而已。

就详细实现而言，它可能大致如下所示：（使用 RAII 以保证干净的簿记）

// Which state the program is in.
enum class MyState {
  idle,
  a,
  b
};
MyState current_state = MyState::idle;

// How many threads of either a or b are holding the resource.
int current_state_count = 0;

// Protects access to current state and count
std::mutex state_mtx;

// Will be notified whenever the state goes back to idle.
std::condition_variable state_cv;


struct my_res_lock {
  my_res_lock(MyState s) {
    assert(s != MyState::idle);

    // Wait until the current state is either idle, or the one we want
    std::unique_lock<std::mutex> lk(state_mtx);
    state_cv.wait(lk, [s]{return current_state == MyState::idle || current_state == s;});

    // If we were in idle, transition to the desired state.
    current_state = s; 
  
    current_state_count += 1;
  }

  ~my_res_lock() {
    std::lock_guard lk(state_mtx);
    current_state_count -= 1;
    if( current_state_count == 0 ) {
      current_state = MyState::idle;

      // whenever we go back to idle, wake up all waiting threads.
      state_cv.notify_all();
    }
  }
};

// in thread A:
{
  my_res_lock res_lock(MyState::a);

  // do some work!
}

// in thread B:
{
  my_res_lock res_lock(MyState::b);

  // do some work!
}

Answer 2

感谢您的出色回答，我设法让它正常工作，但我无法在评论中发布代码，所以我把它放在这里。

在thread_a的开头

bool a=0;
{
  unique_lock<std::mutex> lck(mtx_gpu);
  cv.wait(lck,[]{return cnt_b==0;});//while(!ready) cv.wait(lck)
  mtx_a.lock_shared();//will block if b holds the lock.+1. will cause deadlock here since mtx_b can also be acquired by thread_b
  if(cnt_a++==0) { //check if node_b threads are not running
    mtx_b.lock();
    a=1;
  }
}

在thread_a的末尾：

mtx_a.unlock_shared();//-1
--cnt_a;
if(a){
  mtx_a.lock(); //block until all node_a threads are done
  mtx_b.unlock(); //can only be unlocked by current thread
  cnt_a=0;
  mtx_a.unlock();
  cv.notify_all();
}

这里的重点是使用condition_variable并等待通知，同时释放其他线程的锁。