Как мне работать с мьютексами в подвижных типах в C ++?

По дизайну std::mutexне подлежит перемещению или копированию. Это означает, что класс, Aсодержащий мьютекс, не получит конструктор перемещения по умолчанию.

Как сделать этот тип Aперемещаемым потокобезопасным способом?

Начнем с небольшого кода:

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_блокировки.

Переместить назначение

Оператор присваивания перемещения значительно сложнее, потому что вы не знаете, обращается ли какой-либо другой поток к левой или правой стороне выражения присваивания. И вообще, вам нужно остерегаться следующего сценария:

// 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>;

Это может быть оптимизация, но не определенно. Вам нужно будет измерить, чтобы определить, так ли это. Но с этим изменением можно копировать конструкцию из одной и той же правой руки в несколько потоков одновременно. Решение 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может дать вам представление об API, который может потребоваться для обеспечения «поточно-ориентированного» A, который в целом будет отличаться от «небезопасного для потоков» API из-за проблемы «блокирующей гранулярности».

Также обратите внимание на необходимость защиты от «самостоятельной замены». «Самостоятельная замена» не должна применяться. Без самопроверки можно было бы рекурсивно заблокировать один и тот же мьютекс. Эту проблему также можно решить без самопроверки, используя std::recursive_mutexfor MutexType.

Обновить

В комментариях ниже Якк очень недоволен тем, что ему приходится создавать объекты по умолчанию в конструкторах копирования и перемещения (и он прав). Если вы достаточно серьезно относитесь к этой проблеме, настолько, что готовы потратить на нее память, вы можете избежать этого следующим образом:

• Добавьте любые типы блокировок, которые вам нужны в качестве членов данных. Эти члены должны предшествовать защищаемым данным:

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

К сожалению, Якк стер свой комментарий до того, как я успел завершить это обновление. Но он заслуживает похвалы за то, что поднял эту проблему и нашел решение в этом ответе.

Обновление 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_)
    {}

Учитывая, что, похоже, нет хорошего, чистого и простого способа ответить на это - решение Антона, которое я считаю правильным, но определенно спорным, если не появится лучший ответ, я бы рекомендовал поместить такой класс в кучу и присмотреть за ним через std::unique_ptr:

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

Его теперь полностью подвижный тип и любой, кто имеет замок на внутренней взаимной блокировки в то время как движение происходит по-прежнему безопасно, даже если его спорны ли это хорошая вещь, чтобы сделать

Если вам нужна семантика копирования, просто используйте

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

Это перевернутый ответ. Вместо того, чтобы встраивать «эти объекты должны быть синхронизированы» в качестве основы типа, вставьте его под любой тип.

Вы имеете дело с синхронизированным объектом совсем по-другому. Одна большая проблема - вам нужно беспокоиться о взаимоблокировках (блокировке нескольких объектов). Он также никогда не должен быть вашей «версией объекта по умолчанию»: синхронизированные объекты предназначены для объектов, которые будут конкурировать, и ваша цель должна состоять в том, чтобы минимизировать конкуренцию между потоками, а не скрывать ее.

Но синхронизация объектов по-прежнему полезна. Вместо наследования от синхронизатора мы можем написать класс, который объединяет произвольный тип в синхронизацию. Пользователи должны перепрыгнуть через несколько обручей, чтобы выполнить операции с объектом теперь, когда он синхронизирован, но они не ограничены каким-то ограниченным набором вручную закодированных операций с объектом. Они могут объединить несколько операций над объектом в одну или выполнить операцию над несколькими объектами.

Вот синхронизированная оболочка вокруг произвольного типа 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как многопоточность и записывает его в однопоточном режиме.

В редких случаях, когда нам нужен синхронизированный экземпляр, мы прыгаем через обручи, как показано выше.

Приносим извинения за возможные опечатки в вышеуказанном. Наверное, есть.

Дополнительным преимуществом вышеизложенного является то, что n-арные произвольные операции с synchronizedобъектами (одного и того же типа) работают вместе, без необходимости их жесткого кодирования заранее. Добавьте объявление друга, и n-арные synchronizedобъекты нескольких типов могут работать вместе. В accessтаком случае мне, возможно, придется перестать быть постоянным другом, чтобы иметь дело с конфликтами перегрузки.

живой пример

Использование мьютексов и семантики перемещения C ++ - отличный способ безопасно и эффективно передавать данные между потоками.

Представьте себе поток «производителя», который создает пакеты строк и предоставляет их (одному или нескольким) потребителям. Эти пакеты могут быть представлены объектом, содержащим (потенциально большие) std::vector<std::string>объекты. Мы абсолютно хотим «переместить» внутреннее состояние этих векторов в их потребителей без ненужного дублирования.

Вы просто узнаете мьютекс как часть объекта, а не как часть состояния объекта. То есть вы не хотите перемещать мьютекс.

Какая блокировка вам нужна, зависит от вашего алгоритма или от того, насколько обобщены ваши объекты и какой диапазон использования вы разрешаете.

Если вы только когда - либо перейти от общего состояния «производителя» объекта к локальному потоку «потребляя» объект , который вы могли бы быть в порядке , чтобы только зафиксировать двигающийся от объекта.

Если это более общий дизайн, вам нужно будет заблокировать оба. В таком случае вам нужно подумать о мертвой блокировке.

Если это потенциальная проблема, используйте std::lock()для получения блокировок на обоих мьютексах без взаимоблокировок.

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

В заключение вам необходимо убедиться, что вы понимаете семантику перемещения. Напомним, что перемещенный объект остается в допустимом, но неизвестном состоянии. Вполне возможно, что у потока, не выполняющего перемещение, есть веская причина для попытки доступа к перемещенному объекту, когда он может найти это допустимое, но неизвестное состояние.

Опять же, мой продюсер просто перебирает струны, а потребитель снимает с себя всю нагрузку. В этом случае каждый раз, когда производитель пытается добавить к вектору, он может найти вектор непустым или пустым.

Короче говоря, если потенциальный одновременный доступ к перемещаемому объекту составляет запись, скорее всего, все в порядке. Если это означает чтение, подумайте, почему нормально читать произвольное состояние.

Прежде всего, если вы хотите переместить объект, содержащий мьютекс, что-то не так с вашим дизайном.

Но если вы все равно решите это сделать, вам нужно создать новый мьютекс в конструкторе перемещения, например:

// movable
struct B{};

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

Это потокобезопасный, потому что конструктор перемещения может безопасно предполагать, что его аргумент больше нигде не используется, поэтому блокировка аргумента не требуется.