Проектирование поточно-ориентированного копируемого класса

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

Проектирование потоков происходит на уровне приложений, а не для каждого класса. Только определенные классы управления ресурсами должны иметь потокобезопасность на этом уровне, и для них вам все равно нужно писать явные конструкторы копирования/операторы присваивания.

Вы можете определить свой собственный конструктор копирования (и оператор присваивания копии). Конструктор копирования, вероятно, будет выглядеть примерно так:

cMyClass(const cMyClass& x) : A(x.getA()) { }

Обратите внимание, что getA() должен быть квалифицирован как const, чтобы это работало, что означает, что мьютекс должен быть mutable; вы можете сделать параметр неконстантной ссылкой, но тогда вы не сможете копировать временные объекты, что обычно нежелательно.

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

Каким бы простым ни был вопрос, ответить на него не так-то просто. Для начала мы можем работать с конструктором простого копирования:

// almost pseudo code, mutex/lock/data types are synthetic
class test {
   mutable mutex m;
   data d;
public:
   test( test const & rhs ) {
      lock l(m);         // Lock the rhs to avoid race conditions,
                         // no need to lock this object.
      d = rhs.d;         // perform the copy, data might be many members
   }
};

Теперь создать оператор присваивания стало сложнее. Первое, что приходит на ум, это просто сделать то же самое, но в данном случае заблокировав и левый, и правый:

class test { // wrong
   mutable mutex m;
   data d;
public:
   test( test const & );
   test& operator=( test const & rhs ) {
      lock l1( m );
      lock l2( rhs.m );
      d = rhs.d;
      return *this;
   }
};

Достаточно просто и неправильно. Хотя мы гарантируем однопоточный доступ к объектам (обоим) во время операции и, таким образом, не получаем условий гонки, у нас есть потенциальная взаимоблокировка:

test a, b;
// thr1              // thr2
void foo() {         void bar() {
   a = b;               b = a;
}                    }

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

test& test::operator=( test const & rhs ) {
   if ( this == &rhs ) return *this; // nothing to do
   // same (invalid) code here
}

Что касается другой части проблемы, вам необходимо обеспечить соблюдение порядка получения мьютексов. Это может быть обработано по-разному (сохранение уникального идентификатора для каждого объекта и сравнение...)

test & test::operator=( test const & rhs ) {
   mutex *first, *second;
   if ( unique_id(*this) < unique_id(rhs ) {
      first = &m;
      second = &rhs.m;
   } else {
      first = &rhs.m;
      second = &rhs.m;
   }
   lock l1( *first );
   lock l2( *second );
   d = rhs.d;
}

Конкретный порядок не так важен, как тот факт, что вам нужно обеспечить один и тот же порядок во всех случаях использования, иначе вы потенциально можете заблокировать потоки. Поскольку это довольно распространено, некоторые библиотеки (включая будущий стандарт С++) имеют для него специальную поддержку:

class test {
   mutable std::mutex m;
   data d;
public:
   test( const test & );
   test& operator=( test const & rhs ) {
      if ( this == &rhs ) return *this;        // avoid self deadlock
      std::lock( m, rhs.m );                   // acquire both mutexes or wait
      std::lock_guard<std::mutex> l1( m, std::adopt_lock );      // use RAII to release locks
      std::lock_guard<std::mutex> l2( rhs.m, std::adopt_lock );
      d = rhs.d;
      return *this;
   }
};

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

Это можно записать по-другому, используя std::unique_lock вместо std::lock_guard:

std::unique_lock<std::mutex> l1( m, std::defer_lock );     // store in RAII, but do not lock
std::unique_lock<std::mutex> l2( rhs.m, std::defer_lock );
std::lock( l1, l2 );                                       // acquire the locks

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

test& test::operator=( test copy ) // pass by value!
{
   lock l(m);
   swap( d, copy.d );   // swap is not thread safe
   return *this;
}

}

В обоих подходах есть семантическая разница, так как тот, который использует идиому копирования и замены, имеет потенциальное состояние гонки (которое может повлиять или не повлиять на ваше приложение, но вы должны знать об этом). Поскольку обе блокировки никогда не удерживаются одновременно, объекты могут измениться между моментом снятия первой блокировки (завершение копирования аргумента) и получением второй блокировки внутри operator=.

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

test a(0), b(0); // ommited constructor that initializes the ints to the value
// Thr1
void loop() { // [1]
   while (true) {
      std::unique_lock<std::mutex> la( a.m, std::defer_lock );
      std::unique_lock<std::mutex> lb( b.m, std::defer_lock );
      std::lock( la, lb );
      ++a.d;
      ++b.d;
   }
}
// Thr1
void loop2() {
   while (true) {
      a = b; // [2]
   }
}
// [1] for the sake of simplicity, assume that this is a friend 
//     and has access to members

С реализациями operator=, которые выполняют одновременные блокировки обоих объектов, вы можете утверждать в любой момент времени (делая это безопасно для потоков, получая обе блокировки), что a и b одинаковы, что кажется ожидаемым при беглом чтении код. Это не так, если operator= реализовано с точки зрения идиомы копирования и замены. Дело в том, что в строке, помеченной как [2], b блокируется и копируется во временную, после чего блокировка снимается. Затем первый поток может получить обе блокировки одновременно и увеличить как a, так и b до того, как a будет заблокирован вторым потоком в [2]. Затем a перезаписывается значением, которое b имело до приращения.