如何实现具有自然外观语法的线程安全容器？

Question

前言

下面的代码导致未定义的行为,如果按原样使用:

vector<int> vi;
...
vi.push_back(1);  // thread-1
...
vi.pop(); // thread-2

传统的方法是修复它std::mutex:

std::lock_guard<std::mutex> lock(some_mutex_specifically_for_vi);
vi.push_back(1);

但是,随着代码的增长,这样的事情开始变得麻烦,因为每次在方法之前都会有锁.而且,对于每个对象,我们可能必须保持互斥锁.

目的

在不影响访问对象和声明显式互斥体的语法的情况下,我想创建一个模板,使其完成所有样板工作.例如

Concurrent<vector<int>> vi;  // specific `vi` mutex is auto declared in this wrapper
...
vi.push_back(1); // thread-1: locks `vi` only until `push_back()` is performed
...
vi.pop ()  // thread-2: locks `vi` only until `pop()` is performed

在当前的C++中,实现这一目标是不可能的.不过,我试图在那里,如果只是改变代码vi.来vi->,然后按预期在上面的代码中的注释工作的事情.

码

// The `Class` member is accessed via `->` instead of `.` operator
// For `const` object, it's assumed only for read purpose; hence no mutex lock
template<class Class,
         class Mutex = std::mutex>
class Concurrent : private Class
{
  public: using Class::Class;

  private: class Safe
           {
             public: Safe (Concurrent* const this_,
                           Mutex& rMutex) :
                     m_This(this_),
                     m_rMutex(rMutex)
                     { m_rMutex.lock(); }
             public: ~Safe () { m_rMutex.unlock(); }

             public: Class* operator-> () { return m_This; }
             public: const Class* operator-> () const { return m_This; }
             public: Class& operator* () { return *m_This; }
             public: const Class& operator* () const { return *m_This; }

             private: Concurrent* const m_This;
             private: Mutex& m_rMutex;
           };

  public: Safe ScopeLocked () { return Safe(this, m_Mutex); }
  public: const Class* Unsafe () const { return this; }

  public: Safe operator-> () { return ScopeLocked(); }
  public: const Class* operator-> () const { return this; }
  public: const Class& operator* () const { return *this; }

  private: Mutex m_Mutex;
};

演示

问题

• 是否正在使用临时对象调用具有重载的函数operator->()导致C++中未定义的行为？
• 在所有情况下,这个小实用程序类是否为封装对象提供了线程安全的目的？

---

澄清

对于相互依赖的语句,需要更长的锁定.因此,有一种方法介绍:ScopeLocked().这相当于std::lock_guard().但是,给定对象的互斥锁在内部维护,因此它在语法上仍然更好.
例如,而不是有缺陷的设计(如答案所示):

if(vi->size() > 0)
  i = vi->front(); // Bad: `vi` can change after `size()` & before `front()`

一个人应该依靠以下设计:

auto viLocked = vi.ScopeLocked();
if(viLocked->size() > 0)
  i = viLocked->front();  // OK; `vi` is locked till the scope of `viLocked`

换句话说,对于相互依赖的陈述,应该使用ScopeLocked().

Answer 1

不要这样做.

制作一个线程安全的集合类几乎是不可能的,其中每个方法都会锁定.

请考虑您提出的Concurrent类的以下实例.

Concurrent<vector<int>> vi;

开发人员可能会这样做:

 int result = 0;
 if (vi.size() > 0)
 {
     result = vi.at(0);
 }

另一个线程可能会在第一个线程调用size()和之间进行此更改at(0).

vi.clear();

所以现在,同步的操作顺序是:

vi.size()  // returns 1
vi.clear() // sets the vector's size back to zero
vi.at(0)   // throws exception since size is zero

因此,即使您有一个线程安全的矢量类,两个竞争线程也可能导致在意外的位置抛出异常.

这只是最简单的例子.还有其他方法可以在同一时间尝试读/写/迭代的多个线程可能无意中破坏了线程安全性的保证.

你提到整个事情是由这种繁琐的模式推动的:

vi_mutex.lock();
vi.push_back(1);
vi_mutex.unlock();

实际上,有一些辅助类可以使这个更干净,即lock_guard将使用互斥锁来锁定其构造函数并解析析构函数

{
    lock_guard<mutex> lck(vi_mutex);
    vi.push_back(1);
}

然后在实践中的其他代码变成线程安全ala:

{
     lock_guard<mutex> lck(vi_mutex);
     result = 0;
     if (vi.size() > 0)
     {
         result = vi.at(0);
     }
}

更新:

我编写了一个示例程序,使用您的Concurrent类来演示导致问题的竞争条件.这是代码:

Concurrent<list<int>> g_list;

void thread1()
{
    while (true)
    {
        if (g_list->size() > 0)
        {
            int value = g_list->front();
            cout << value << endl;
        }
    }

}

void thread2()
{
    int i = 0;
    while (true)
    {
        if (i % 2)
        {
            g_list->push_back(i);
        }
        else
        {
            g_list->clear();
        }
        i++;
    }
}

int main()
{

    std::thread t1(thread1);
    std::thread t2(thread2);

    t1.join(); // run forever

    return 0;
}

在非优化版本中,上面的程序会在几秒钟内崩溃.(零售业有点困难,但问题仍然存在).

Answer 2

这种努力充满了危险和性能问题.迭代器通常依赖于整个数据结构的状态,并且如果数据结构以某种方式改变,则通常会失效.这意味着迭代器要么在创建它们时需要在整个数据结构上保存互斥锁,要么你需要定义一个特殊的迭代器,它只是小心地锁定它所依赖的东西,这可能不仅仅是它当前指向的节点/元素的状态.这需要内部了解所包含内容的实现.

举个例子,想想这个事件序列可能会如何发挥作用:

线程1:

 void thread1_func(Concurrent<vector<int>> &cq)
 {
       cq.push_back(1);
       cq.push_back(2);
 }

线程2:

 void thread2_func(Concurrent<vector<int>> &cq)
 {
       ::std::copy(cq.begin(), cq.end(), ostream_iterator<int>(cout, ", "));
 }

你怎么认为这会发挥出来？即使每个成员函数都很好地包装在一个互斥锁中,所以它们都是序列化和原子的,你仍然会调用未定义的行为,因为一个线程改变了另一个迭代的数据结构.

您可以创建迭代器也锁定互斥锁.但是,如果同一个线程创建另一个迭代器,它应该能够获取互斥锁,因此您需要使用递归互斥锁.

当然,这意味着当一个线程在其上进行迭代时,任何其他线程都无法触及您的数据结构,从而显着降低了并发机会.

它也很容易出现竞争条件.一个线程进行调用并发现它感兴趣的数据结构的一些事实.然后,假设这个事实是真的,它会再次调用.但是,当然,事实已经不再是真实的了,因为其他一些线索在获取事实和使用事实之间捅了它的鼻子.使用size然后决定是否迭代它的例子只是一个例子.

Answer 3

使用临时对象调用具有重载运算符的函数 - >()导致C++中的未定义行为

不,临时工作只有在完全表达结束时才会被摧毁,使他们恢复活力.并且使用具有重载的临时对象operator->来"装饰"成员访问正是为什么重载运算符按照它的方式定义的原因.它用于在专用构建中进行日志记录,性能测量,并且像您自己发现的那样,将所有成员访问锁定到封装对象.

在这种情况下,基于循环语法的范围不起作用.它给出了编译错误.修复它的正确方法是什么？

Iterator据我所知,你的函数不会返回实际的迭代器.Safe<Args...>(std::forward<Args>(args)...);与参数列表比较Iterator(Class::NAME(), m_Mutex).什么Base时候Args推论出来的论点Class::NAME()？

在所有情况下,这个小实用程序类是否为封装对象提供了线程安全的目的？

对于简单的值类型,它看起来相当安全.但当然这取决于通过包装器完成的所有访问.

对于更复杂的容器,考虑迭代器失效,然后使单个成员访问原子不一定会阻止竞争条件(如注释中所述).我想你可能会创建一个迭代器包装器,它在生命周期内锁定容器......但是你失去了大部分有用的容器API.