在C++中，如何处理可移动类型的互斥锁？

Question

按照设计，std::mutex 不可移动也不可复制。这意味着一个拥有互斥锁的类 A 将不会接收默认移动构造函数。

我该如何以线程安全的方式使这种类型 A 可移动？

- Jack Sabbath

---

这个问题有一个小细节：移动操作本身是否也需要线程安全性，或者只是对象的其他访问需要线程安全？ - Jonas Schäfer

@paulm 这实际上取决于设计。我经常看到一个类有一个mutex成员变量，然后只有 std::lock_guard 是方法作用域的。 - Cory Kramer

@Anton Savin：我可以想到很多用例...一个例子是“可锁定”的基类，不应该限制派生类的可移动性。 - Jack Sabbath

@Jonas Wielicki：起初我以为移动它也应该是线程安全的。然而，现在再仔细想一想，这似乎没有多大意义，因为移动构造一个对象通常会使旧对象的状态无效。所以如果要移动对象，其他线程不能访问旧对象，否则它们可能很快就会访问到一个无效的对象。我说得对吗？ - Jack Sabbath

请使用以下链接以获取完整内容，该内容可能对您有所帮助： https://www.justsoftwaresolutions.co.uk/threading/thread-safe-copy-constructors.html。我将为您翻译这篇文章，使其更通俗易懂，但不会改变原意。 - Ravi Chauhan

Answer 1

我们从一些代码开始:

class A
{
    using MutexType = std::mutex;
    using ReadLock = std::unique_lock<MutexType>;
    using WriteLock = std::unique_lock<MutexType>;

    mutable MutexType mut_;

    std::string field1_;
    std::string field2_;

public:
    ...

我在里面放了一些相当暗示性的类型别名，虽然我们在C++11中不会真正利用它们，但在C++14中它们变得更加有用。耐心等待，我们会到那里的。

你的问题可以归结为：

如何为这个类编写移动构造函数和移动赋值运算符？

我们将从移动构造函数开始。

移动构造函数

请注意，成员mutex已经被设置为mutable。严格来说，在移动成员中这并非必需，但我假设你还想要复制成员。如果不是这样的话，则没有必要使互斥量成为mutable。

在构造A时，您不需要锁定this->mut_。但您需要锁定正在构造的对象（移动或复制）的mut_。可以这样做：

    A(A&& a)
    {
        WriteLock rhs_lk(a.mut_);
        field1_ = std::move(a.field1_);
        field2_ = std::move(a.field2_);
    }

请注意，我们首先必须默认构造this的成员，然后仅在a.mut_加锁后再为其赋值。

移动赋值

移动赋值运算符要复杂得多，因为您不知道其他某个线程是否正在访问赋值表达式的lhs或rhs。 一般而言，您需要防范以下情况：

// Thread 1
x = std::move(y);

// Thread 2
y = std::move(x);

这里是正确保护上述场景的移动赋值运算符：

    A& operator=(A&& a)
    {
        if (this != &a)
        {
            WriteLock lhs_lk(mut_, std::defer_lock);
            WriteLock rhs_lk(a.mut_, std::defer_lock);
            std::lock(lhs_lk, rhs_lk);
            field1_ = std::move(a.field1_);
            field2_ = std::move(a.field2_);
        }
        return *this;
    }

请注意，必须使用std::lock(m1, m2)来锁定这两个互斥量，而不是只是一个接一个地锁定它们。如果您一个接一个地锁定它们，那么当两个线程按照上面显示的相反顺序分配两个对象时，您可能会遇到死锁。 std::lock的作用是避免这种死锁。

拷贝构造函数

虽然你没有询问拷贝成员，但我们也可以现在谈论它们（如果不是你，总有人会需要它们）。

    A(const A& a)
    {
        ReadLock  rhs_lk(a.mut_);
        field1_ = a.field1_;
        field2_ = a.field2_;
    }

复制构造函数看起来与移动构造函数很相似，只是使用了ReadLock别名而不是WriteLock。目前这两个别名都是指向std::unique_lock<std::mutex>，因此并没有什么区别。

但在C++14中，您将有以下选项：

    using MutexType = std::shared_timed_mutex;
    using ReadLock  = std::shared_lock<MutexType>;
    using WriteLock = std::unique_lock<MutexType>;

这 可能 是一种优化方式，但并不确定。你需要进行测量来确定是否是优化。但是通过此更改，可以在多个线程中同时从相同的 rhs 进行复制构造。C++11 解决方案会强制您使这些线程顺序执行，即使 rhs 没有被修改。

复制分配

为了完整起见，这里是复制分配运算符，阅读其他内容后应该相当容易理解：

    A& operator=(const A& a)
    {
        if (this != &a)
        {
            WriteLock lhs_lk(mut_, std::defer_lock);
            ReadLock  rhs_lk(a.mut_, std::defer_lock);
            std::lock(lhs_lk, rhs_lk);
            field1_ = a.field1_;
            field2_ = a.field2_;
        }
        return *this;
    }

等等。

如果您希望多个线程能够同时调用访问A状态的任何其他成员或自由函数，那么这些函数也需要受到保护。例如，这里是swap：

    friend void swap(A& x, A& y)
    {
        if (&x != &y)
        {
            WriteLock lhs_lk(x.mut_, std::defer_lock);
            WriteLock rhs_lk(y.mut_, std::defer_lock);
            std::lock(lhs_lk, rhs_lk);
            using std::swap;
            swap(x.field1_, y.field1_);
            swap(x.field2_, y.field2_);
        }
    }

请注意，如果仅依赖于std::swap，则锁定粒度将不正确，在std::swap内部执行的三个移动之间进行锁定和解锁。

确实，思考swap可以让你了解到为“线程安全”的A提供的API可能需要与“非线程安全”的API不同，因为存在“锁定粒度”问题。

还要注意保护自身交换。 "self-swap" 应该是无操作的。没有自检，将递归锁定相同的互斥量。 这也可以通过对MutexType使用std::recursive_mutex来解决而不进行自检。

更新：

在下面的评论中，Yakk对于在复制和移动构造函数中强制默认构造某些内容感到不满（他有一定道理）。 如果您对此问题感到非常强烈，以至于愿意在其上花费内存，那么可以这样避免：

添加所需的任何锁定类型作为数据成员。 这些成员必须出现在被保护的数据之前：

mutable MutexType mut_;
ReadLock  read_lock_;
WriteLock write_lock_;
// ... other data members ...

然后在构造函数中（例如复制构造函数）执行以下操作：

A(const A& a)
    : read_lock_(a.mut_)
    , field1_(a.field1_)
    , field2_(a.field2_)
{
    read_lock_.unlock();
}

哎呀，Yakk在我完成这个更新之前把他的评论删掉了。但是他应该得到推动这个问题并将解决方案纳入这个答案的功劳。

更新2

dyp提出了这个好建议：

    A(const A& a)
        : A(a, ReadLock(a.mut_))
    {}
private:
    A(const A& a, ReadLock rhs_lk)
        : field1_(a.field1_)
        , field2_(a.field2_)
    {}

- Howard Hinnant

---

你的复制构造函数只是赋值字段，而不是复制它们。这意味着它们需要具备默认构造能力，这是一个不幸的限制。 - Yakk - Adam Nevraumont

@Yakk：是的，将“mutexes”放入类类型中并不是“唯一正确的方法”。这只是工具箱中的一个工具，如果您想使用它，就是这样。 - Howard Hinnant

@Yakk：在我的回答中搜索字符串“C++14”。 - Howard Hinnant

啊，抱歉，我漏掉了那个C++14的部分。 - Yakk - Adam Nevraumont

非常感谢 @HowardHinnant 的出色解释！在 C++17 中，您还可以使用 std::scoped_lock lock(x.mut_, y_mut_); 这样，您就可以依赖于实现以正确的顺序锁定多个互斥量。 - fen

Answer 2

鉴于目前似乎没有一种好的、干净的、简单的方法来回答这个问题——我认为Anton的解决方案是正确的，但这肯定是有争议的，除非有更好的答案出现，否则我建议将这样的类放在堆上，并通过 std::unique_ptr 进行管理：

auto a = std::make_unique<A>();

现在它是全可移动类型，任何在移动发生时锁定内部互斥锁的人仍然是安全的，尽管这是否是一个好做法还有争议

如果你需要复制语义，只需使用

auto a2 = std::make_shared<A>();

- Mike Vine

Answer 3

这是一个颠倒的答案。不是将“此对象需要同步”作为类型的基础嵌入，而是在任何类型下注入它。

您需要以非常不同的方式处理同步对象。一个大问题是你必须担心死锁（锁定多个对象）。它也基本上不应该是您“对象的默认版本”，同步对象是用于会争用的对象，您的目标应该是尽量减少线程之间的争用，而不是掩盖它。

但同步对象仍然很有用。我们可以编写一个类，将任意类型包装在同步中，而不是从同步器继承。现在，用户必须跳过一些障碍才能对已同步的对象执行操作，但他们不限于对对象进行一些手工编码的有限操作集。他们可以将多个操作组合成一个对象，或者对多个对象执行操作。

这是一个围绕任意类型T的同步包装器：

template<class T>
struct synchronized {
  template<class F>
  auto read(F&& f) const&->std::result_of_t<F(T const&)> {
    return access(std::forward<F>(f), *this);
  }
  template<class F>
  auto read(F&& f) &&->std::result_of_t<F(T&&)> {
    return access(std::forward<F>(f), std::move(*this));
  }
  template<class F>
  auto write(F&& f)->std::result_of_t<F(T&)> {
    return access(std::forward<F>(f), *this);
  }
  // uses `const` ness of Syncs to determine access:
  template<class F, class... Syncs>
  friend auto access( F&& f, Syncs&&... syncs )->
  std::result_of_t< F(decltype(std::forward<Syncs>(syncs).t)...) >
  {
    return access2( std::index_sequence_for<Syncs...>{}, std::forward<F>(f), std::forward<Syncs>(syncs)... );
  };
  synchronized(synchronized const& o):t(o.read([](T const&o){return o;})){}
  synchronized(synchronized && o):t(std::move(o).read([](T&&o){return std::move(o);})){}  
  // special member functions:
  synchronized( T & o ):t(o) {}
  synchronized( T const& o ):t(o) {}
  synchronized( T && o ):t(std::move(o)) {}
  synchronized( T const&& o ):t(std::move(o)) {}
  synchronized& operator=(T const& o) {
    write([&](T& t){
      t=o;
    });
    return *this;
  }
  synchronized& operator=(T && o) {
    write([&](T& t){
      t=std::move(o);
    });
    return *this;
  }
private:
  template<class X, class S>
  static auto smart_lock(S const& s) {
    return std::shared_lock< std::shared_timed_mutex >(s.m, X{});
  }
  template<class X, class S>
  static auto smart_lock(S& s) {
    return std::unique_lock< std::shared_timed_mutex >(s.m, X{});
  }
  template<class L>
  static void lock(L& lockable) {
      lockable.lock();
  }
  template<class...Ls>
  static void lock(Ls&... lockable) {
      std::lock( lockable... );
  }
  template<size_t...Is, class F, class...Syncs>
  friend auto access2( std::index_sequence<Is...>, F&&f, Syncs&&...syncs)->
  std::result_of_t< F(decltype(std::forward<Syncs>(syncs).t)...) >
  {
    auto locks = std::make_tuple( smart_lock<std::defer_lock_t>(syncs)... );
    lock( std::get<Is>(locks)... );
    return std::forward<F>(f)(std::forward<Syncs>(syncs).t ...);
  }

  mutable std::shared_timed_mutex m;
  T t;
};
template<class T>
synchronized< T > sync( T&& t ) {
  return {std::forward<T>(t)};
}

包含C++14和C++1z的特性。

这假设const操作是多读安全的（这是std容器所假定的）。

使用方法如下：

synchronized<int> x = 7;
x.read([&](auto&& v){
  std::cout << v << '\n';
});

针对具有同步访问的int，我建议不要使用synchronized(synchronized const&)。这很少需要。

如果您需要synchronized(synchronized const&)，我倾向于使用std::aligned_storage替换T t;，允许手动放置构造，并进行手动销毁。这允许正确的生命周期管理。

除此之外，我们可以复制源T，然后从中读取：

synchronized(synchronized const& o):
  t(o.read(
    [](T const&o){return o;})
  )
{}
synchronized(synchronized && o):
  t(std::move(o).read(
    [](T&&o){return std::move(o);})
  )
{}

用于赋值：

synchronized& operator=(synchronized const& o) {
  access([](T& lhs, T const& rhs){
    lhs = rhs;
  }, *this, o);
  return *this;
}
synchronized& operator=(synchronized && o) {
  access([](T& lhs, T&& rhs){
    lhs = std::move(rhs);
  }, *this, std::move(o));
  return *this;
}
friend void swap(synchronized& lhs, synchronized& rhs) {
  access([](T& lhs, T& rhs){
    using std::swap;
    swap(lhs, rhs);
  }, *this, o);
}

放置和对齐存储版本有点混乱。大多数对的访问将被成员函数 T()和T const() const取代，除了在构造时需要跳过一些障碍。

通过使“synchronized”成为包装器而不是类的一部分，我们只需确保类内部将“const”视为多读取器，并以单线程方式编写它。

在我们需要同步实例的罕见情况下，我们要像上面那样跳过一些障碍。

对于上面的任何打字错误，我表示歉意。很可能有一些。

上述方法的一个附带好处是，synchronized对象（相同类型）的n元任意操作可以一起工作，而无需事先硬编码。添加友元声明并且多种类型的n元对象可能会一起工作。在这种情况下，我可能不得不将移出成为内联朋友，以处理重载冲突。

{{链接1：实时示例}}

- Yakk - Adam Nevraumont

Answer 4

首先，如果你想移动一个包含互斥锁的对象，那么你的设计肯定有问题。

但如果你决定这样做，你需要在移动构造函数中创建一个新的互斥锁，例如：

// movable
struct B{};

class A {
    B b;
    std::mutex m;
public:
    A(A&& a)
        : b(std::move(a.b))
        // m is default-initialized.
    {
    }
};

这是线程安全的，因为移动构造函数可以安全地假设其参数不会在任何其他地方使用，因此不需要锁定参数。

- Anton Savin

---

为什么将包含互斥锁的可移动类型视为不良设计？我认为从工厂函数中返回线程安全对象是一个好主意。将线程安全对象移动到向量中对我来说也很合适。 - Jack Sabbath

这不是线程安全的。如果a.mutex被锁定，你会失去那个状态。-1 - user2249683

只要参数是对被移动对象的唯一引用，那么就没有理智的理由去锁定它的互斥量。即使如此，也没有理由去锁定一个新创建对象的互斥量。如果真的有这样的需要，那么这说明可移动对象与互斥量的整体设计存在问题。 - Anton Savin

@DieterLücking 移动后对象的互斥保护在移动构造函数执行后就不再保护任何内容。而且，该移动后对象即将被销毁。那么，创建一个新互斥体到底是如何引入任何线程不安全性的呢？ - Anton Savin

然而，如果这是最好的方法，我仍会建议创建一个实例并将其放置在std::unique_ptr中 - 这似乎更简洁，不太可能导致混淆问题。好问题。 - Mike Vine

Answer 5

使用互斥锁和C++移动语义是一种安全高效的在线程之间传输数据的方法。

想象一个“生产者”线程制作字符串批次并将它们提供给（一个或多个）消费者。这些批次可以由包含（潜在的大型）std::vector<std::string>对象的对象表示。 我们绝对希望将这些向量的内部状态“移动”到它们的消费者，而不需要不必要的复制。

你只需将互斥锁视为对象的一部分而不是对象状态的一部分即可。也就是说，您不想移动互斥锁。

您需要的锁定取决于您的算法或对象的通用程度以及您允许的使用范围。

如果您仅从共享状态的“生产者”对象移动到线程本地的“消费”对象，则可能只需锁定已移动的from对象。

如果这是一个更普遍的设计，您将需要锁定两个对象。在这种情况下，您需要考虑死锁问题。

如果存在潜在问题，则使用std::lock()以无死锁方式获取两个互斥锁。

http://en.cppreference.com/w/cpp/thread/lock

作为最后的说明，您需要确保理解移动语义。 请注意，被移动的对象处于有效但未知状态。 完全有可能，当没有执行移动的线程尝试访问已被移动的对象时，它可能会发现该对象处于有效但未知状态。

再次提醒，我的生产者只是不断地产生字符串，而消费者则将整个负载拿走。在这种情况下，每次生产者尝试添加到向量中时，它可能会发现向量非空或为空。

简而言之，如果对已移动对象的潜在并发访问涉及写入，则很可能是可以的。如果涉及读取，则需要考虑为什么可以读取任意状态。

- Persixty