我应该如何处理C++中可移动类型的互斥量？

Question

按设计,std::mutex不可移动也不可复制.这意味着A拥有互斥锁的类不会接收default-move-constructor.

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

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_);
    }

请注意,我们必须默认构造thisfirst 的成员,然后仅在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的构造.即使没有修改rhs,C++ 11解决方案也会强制您按顺序创建此类线程.

复制分配

为了完整起见,这里是复制赋值运算符,在阅读其他所有内容之后,它应该是相当自我解释的:

    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不同.

还要注意防止"自我交换"的必要性."自我交换"应该是无操作.如果没有自检,则会递归锁定相同的互斥锁.这也可以通过使用std::recursive_mutexfor 而无需自我检查来解决MutexType.

更新

在下面的评论中,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_)
    {}

Answer 2

鉴于似乎没有一个好的,干净的,简单的方法来回答这个问题 - 安东的解决方案我认为是正确的,但它肯定存在争议,除非出现更好的答案,我建议将这样的课程放在堆上并照顾它通过std::unique_ptr:

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

它现在是一个完全可移动的类型,任何在内部互斥锁上锁定同时移动发生的人仍然是安全的,即使它有争议这是否是一件好事

如果您需要复制语义,请使用

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

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&),我会忍不住来代替T t;用std::aligned_storage,允许手动放置建设,做手工破坏.这允许适当的寿命管理.

除此之外,我们可以复制源代码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&t()和T const&t()const,除了在建筑,你就必须通过一些跳铁圈.

通过创建synchronized包装器而不是类的一部分,我们必须确保的是,类在内部尊重const多读取器,并以单线程方式编写它.

在极少数情况下,我们需要一个同步的实例,我们跳过如上所述的箍.

对上述任何拼写错误道歉.可能有一些.

上述的另一个好处是对synchronized对象(相同类型)的n元任意操作可以一起工作,而无需事先对其进行硬编码.添加好友声明和synchronized多种类型的n-ary 对象可以一起使用.access在这种情况下,我可能不得不离开成为内联朋友来处理过载问题.

实例