将互斥保护构建到 C++ 类中的线程安全方法？

Posted 2023-02-16

【中文标题】将互斥保护构建到 C++ 类中的线程安全方法？

【英文标题】：Thread-safe way to build mutex protection into a C++ class?

【发布时间】：2012-07-24 23:19:52

【问题描述】：

我正在尝试在 C++ 中为我正在进行的项目实现生产者/消费者模型多线程程序。基本思想是主线程创建第二个线程来监视串行端口的新数据，处理数据并将结果放入由主线程定期轮询的缓冲区中。我以前从未编写过多线程程序。我一直在阅读很多教程，但它们都是用 C 语言编写的。我想我已经掌握了基本概念，但我正在尝试对其进行 c++ 化。对于缓冲区，我想创建一个内置互斥保护的数据类。这就是我想出的。

1) 我是不是走错了路？有没有更聪明的方法来实现受保护的数据类？

2) 如果两个线程同时尝试调用ProtectedBuffer::add_back()，下面的代码会发生什么？

#include <deque>
#include "pthread.h"

template <class T>
class ProtectedBuffer 
  std::deque<T> buffer;
  pthread_mutex_t mutex;
public:
  void add_back(T data) 
    pthread_mutex_lock(&mutex);
    buffer.push_back(data);
    pthread_mutex_unlock(&mutex);
  
  void get_front(T &data) 
    pthread_mutex_lock(&mutex);
    data = buffer.front();
    buffer.pop_front();
    pthread_mutex_unlock(&mutex);
  
;

编辑： 感谢所有伟大的建议。我试图在下面实现它们。我还添加了一些错误检查，所以如果一个线程以某种方式设法尝试锁定同一个互斥锁两次，它将优雅地失败。我想。

#include "pthread.h"
#include <deque>


class Lock 
    pthread_mutex_t &m;
    bool locked;
    int error;
public:
    explicit Lock(pthread_mutex_t & _m) : m(_m) 
        error = pthread_mutex_lock(&m);
        if (error == 0) 
            locked = true;
         else 
            locked = false;
        
    
    ~Lock() 
        if (locked)
            pthread_mutex_unlock(&m);
    
    bool is_locked() 
        return locked;
    
;

class TryToLock 
    pthread_mutex_t &m;
    bool locked;
    int error;
public:
    explicit TryToLock(pthread_mutex_t & _m) : m(_m) 
        error = pthread_mutex_trylock(&m);
        if (error == 0) 
            locked = true;
         else 
            locked = false;
        
    
    ~TryToLock() 
        if (locked)
            pthread_mutex_unlock(&m);
    
    bool is_locked() 
        return locked;
    
;

template <class T>
class ProtectedBuffer
    pthread_mutex_t mutex;
    pthread_mutexattr_t mattr;
    std::deque<T> buffer;
    bool failbit;

    ProtectedBuffer(const ProtectedBuffer& x);
    ProtectedBuffer& operator= (const ProtectedBuffer& x);
public:
    ProtectedBuffer() 
        pthread_mutexattr_init(&mattr);
        pthread_mutexattr_settype(&mattr, PTHREAD_MUTEX_ERRORCHECK);
        pthread_mutex_init(&mutex, &mattr);
        failbit = false;
    
    ~ProtectedBuffer() 
        pthread_mutex_destroy(&mutex);
        pthread_mutexattr_destroy(&mattr);
    
    void add_back(T &data) 
        Lock lck(mutex);
        if (!lck.locked()) 
            failbit = true;
            return;
        
        buffer.push_back(data);
        failbit = false;
    
    void get_front(T &data) 
        Lock lck(mutex);
        if (!lck.locked()) 
            failbit = true;
            return;
        
        if (buffer.empty()) 
            failbit = true;
            return;
        
        data = buffer.front();
        buffer.pop_front();
        failbit = false;
    
    void try_get_front(T &data) 
        TryToLock lck(mutex);
        if (!lck.locked()) 
            failbit = true;
            return;
        
        if (buffer.empty()) 
            failbit = true;
            return;
        
        data = buffer.front();
        buffer.pop_front();
        failbit = false;
    
    void try_add_back(T &data) 
        TryToLock lck(mutex);
        if (!lck.locked()) 
            failbit = true;
            return;
        
        buffer.push_back(data);
        failbit = false;
    
;

【问题讨论】：

thegeekstuff.com/2012/07/c-thread-safe-and-reentrant

【参考方案1】：

几件事：

您需要在构造函数中使用pthread_mutex_init 初始化mutex，并在析构函数中使用pthread_mutex_destroy 释放它。

您必须使您的类不可复制和不可赋值（或以其他方式正确实现复制构造函数和赋值运算符；见上文）。

为锁创建一个 SBRM 帮助类是值得的：

class Lock

    pthread_mutex_t & m;
public:
    explicit Lock(pthread_mutex_t & _m) : m(_m)  pthread_mutex_lock(&m); 
    ~Lock()  pthread_mutex_unlock(&m); 
;

现在您可以创建一个同步范围，例如 Lock lk(mutex); /* ... */ 。

至于问题2：并发访问是通过锁定互斥体的方式进行序列化的。其中一个竞争线程将在获取互斥锁时休眠。

【讨论】：

我想我理解助手类。几个问题。一、pthread_mutex_t & m;中的&是怎么回事？第二，如果我想要一个非阻塞的class Lock，我是否只需将pthread_mutex_lock(&m) 替换为pthread_mutex_try_lock(&m) 并让我的助手类在它返回为“忙碌”时抛出异常？

@Beezum: One) 这是一个参考； 2) try-lock 不能很好地与作用域资源管理习语混合；您必须创建一个更复杂的类，该类可以解锁构造，然后以某种方式公开 try-lock 调用并返回一个布尔值。使用 try-lock 的代码更复杂。

顺便问一下什么是SBRM？

@DhanarajDurairaj：“基于范围的资源管理”，即使用析构函数和自动变量来管理资源。有时称为“RAII”，但这不是一个好词。

【参考方案2】：

我是不是走错了路？有没有更聪明的方法来实现受保护的数据类？

对于您的实施，我认为您有一个良好的开端。既然你问的是 C++ifying，那么如果你有一个支持 C++11 的编译器，你可以使用新的线程支持。

您提到您希望主线程轮询此缓冲区，但我没有看到任何允许它这样做的机制。 get_front 应该在缓冲区中没有任何内容时提供错误，或者get_buffer 应该阻止调用者直到数据可用。

#include <deque>
#include <mutex>
#include <condition_variable>
#include <stdexcept>

template <class T>
class ProtectedBuffer 
  std::deque<T> buffer;
  std::mutex mtx;
  std::condition_variable empty_cnd;
  void get_front_i(T &data) 
    data = buffer.front();
    buffer.pop_front();
  
public:
  void add_back(T data) 
    std::lock_guard<std::mutex> g(mtx);
    bool was_empty = buffer.empty();
    buffer.push_back(data);
    if (was_empty) empty_cnd.notify_one();
  
  void get_front_check(T &data) 
    std::lock_guard<std::mutex> g(mtx);
    if (buffer.empty()) throw std::underflow_error("no data");
    get_front_i(data);
  
  void get_front_block(T &data) 
    std::lock_guard<std::mutex> g(mtx);
    std::unique_lock<std::mutex> u(mtx);
    while (buffer.empty()) empty_cnd.wait(u);
    get_front_i(data);
    if (!buffer.empty()) empty_cnd.notify_one();
  
;

如果您想限制添加到缓冲区的数据量，您可以添加一个类似的full_cnd 条件变量来检查add_back 调用将等待的完整条件（如果为真）。然后，get_front_i 方法可以在缓冲区不再满时发出信号。

如果两个线程尝试同时调用 ProtectedBuffer::add_back()，下面的代码会发生什么？

由于add_back受到互斥保护，如果两个线程同时调用它，一个线程将被阻止调用push_back，直到另一个线程完成。

【参考方案3】：

您有基础知识，但我会更进一步，将互斥锁本身包装在其自己的 RAII 包装器中，例如：

#include <deque> 
#include "pthread.h" 

class ProtectedMutex

  pthread_mutex_t &mutex; 
public:
  ProtectedMutex(pthread_mutex_t &m)
    : mutex(m); 
  
    pthread_mutex_lock(&mutex); 
  
  ~ProtectedMutex()
  
    pthread_mutex_unlock(&mutex); 
  
;

template <class T> 
class ProtectedBuffer  
  std::deque<T> buffer; 
  pthread_mutex_t mutex; 
public: 
  void add_back(T data)  
    ProtectedMutex m(mutex); 
    buffer.push_back(data); 
   
  void get_front(T &data)  
    ProtectedMutex m(mutex); 
    data = buffer.front(); 
    buffer.pop_front(); 
   
; 

【参考方案4】：

' 将结果放入由主线程定期轮询的缓冲区中' - CPU 浪费和延迟。

“我是不是走错路了？” - 是的。我不知道您的系统对辅助线程 GUI 线程通信有什么样的支持，但总有 PostMessage() API。

当然，您需要一个 Buffer 类，其中包含用于串行 rx 数据的数据成员和执行协议/“处理数据”的方法。你不需要太多其他东西。在您的第二个线程中，创建一个缓冲区类实例。加载它，处理数据和 PostMessage/dispatch/BeginInvoke 它指向您的 GUI 线程的指针。在串行线程的下一行代码中，在同一个实例指针 var 中创建另一个实例，以便从串行端口加载下一次数据。在 GUI 中显示/记录/任何内容之后，GUI 线程应该删除（）它收到的 *Buffer。

没有延迟，没有 CPU 浪费，没有数据复制，串行线程和 GUI 线程不可能在同一个缓冲区实例上工作，没有讨厌、复杂的缓冲区共享代码，没有锁，没有麻烦。它会很好用。

其他任何事情都会变得一团糟。

编辑 - 忘记了 (2) - 不知道，。不会用驳船杆碰它..

【讨论】：

在 Windows 之外没有 PostMessage API，由于他的问题引用了 pthread_mutex_t，他可能没有使用它...在类 Unix 系统中执行此操作的方法是使用 @987654323 @ 或其替代品之一（epoll、kqueue 等）

我应该提到这将进入嵌入式 linux 系统。我根据@asveikau 的建议查找了epoll。我对如何在这种情况下使用它感到有些困惑。我是否必须在文件系统中创建一个新文件来向主线程发出信号？这会比使用条件变量更好吗？我考虑过条件变量，但我不希望我的主线程在等待条件变量时阻塞。在串行接口工作时，它还有其他事情要处理。

@Beezum - 关键的见解是，最终，您的串行端口是一个文件（并且具有可以轮询的文件描述符）。我不确定你的主线程做了什么，但如果它使用 X UI 或终端接口，它可能还会阻塞来自文件描述符的事件......对poll 或epoll 的调用循环可以使你的进程继续进行休眠，直到它从这些来源中的任何一个接收到一个事件。