Делаем любой объект потокобезопасным

В этих 3-ех статьях я детально расскажу об атомарных операциях, барьерах памяти и о быстром обмене данными между потоками, а так же о «sequence-points» на примере «execute-around-idiom», а заодно постараемся вместе сделать что-нибудь полезное — умный указатель, который делает любой объект потоко-безопасным для любых операций с его членами переменными или функциями. А затем покажем как используя его достичь производительности высоко-оптимизированных lock-free алгоритмов на 8 — 64 ядрах.

Три связанные статьи:

1. Делаем любой объект потокобезопасным
2. Ускоряем std::shared_mutex в 10 раз
3. Потокобезопасный std::map с производительностью lock-free map

Моя статья на английском: www.codeproject.com/Articles/1183379/We-make-any-object-thread-safe
Примеры и тесты из всех трех статей: github.com/AlexeyAB/object_threadsafe

В стандартной библиотеке C++ не существует потоко-безопасных контейнеров (array, list, map…), которые можно использовать в нескольких потоках без дополнительных блокировок. Используя стандартные контейнеры для многопоточного обмена есть опасность, что вы забудете защитить один из участков кода mutex-ом или по ошибке защитите его другим mutex-ом.
Очевидно, что разработчик совершит больше ошибок если будет использовать собственное решение, вместо стандартного. А если задача настолько сложная, что решения нет в стандарте, то разработчик решая её утонет в ошибках.
Исходя из принципа «практичность важнее безупречности» («practicality beats purity»), мы постараемся создать не идеальное, но практичное решение этой задачи.
В статье мы реализуем умный указатель, который делает любой объект потоко-безопасным, с производительностью не уступающей оптимизированным lock-free контейнерам.
Упрощенный неоптимизированный пример использования такого указателя:

int main() {    
    contfree_safe_ptr< std::map<std::string, int> > safe_map_string_int;
    
    std::thread t1([&]() { safe_map_string_int->emplace("apple", 1); }); 
    std::thread t2([&]() { safe_map_string_int->emplace("potato", 2); }); 
    t1.join(); t2.join();

    std::cout << "apple = " << (*safe_map_string_int)["apple"] << 
        ", potato = " << (*safe_map_string_int)["potato"] << std::endl;    
    return 0;
}

На каждый шаг нашего алгоритма мы приведем цитаты из стандарта гарантирующие необходимое поведение.
Мы детально рассмотрим модель памяти C++, разные возможные варианты синхронизации и обмена данными между потоками.
Во второй статье мы реализуем оптимизированный «contention-free shared-mutex» и оптимизированный указатель на его основе contfree_safe_ptr<>. А в третей статье покажем примеры оптимального использования contfree_safe_ptr<> и приведем тесты производительности.

Начало

Начнем с разработки шаблона умного указателя safe_ptr<T>, который будет потоко-безопасен для любого типа T и покажет многопоточную производительность на уровне оптимизированных lock-free алгоритмов.
Причем, с возможностью работать атомарно и консистентно сразу над несколькими разными объектами, где даже для lock-free структур данных пришлось бы использовать блокировки и появился бы риск вечных deadlock. Но мы разработаем специальный класс блокировки мьютекса разрешающий ситуации deadlocks. Затем реализуем собственный высокопроизводительный contention-free shared-mutex, который значительно быстрее стандартного std::shared_mutex. И на его основе создадим оптимизированную версию безопасного указателя safe_ptr<T>, названную contfree_safe_ptr<>. В конце проведем тесты производительности сравнивая с lock-free библиотекой CDSlib. Мы увидим близкую производительность к CDSlib (SkipListMap и BronsonAVLTreeMap) на примере contfree_safe_ptr<std::map<>>.
В результате, чтобы сделать любой ваш класс T потоко-безопасным от вас не потребуется времени, просто пишите: contfree_safe_ptr<T> ptr_thread_safe;
А производительность окажется близкой к тому, как если бы вы месяц разрабатывали lock-free алгоритмы в методах вашего класса. К тому же будет возможность атомарно изменять сразу несколько contfree_safe_ptr<>. Так же как и std::shared_ptr<> наш умный указатель будет содержать счетчик ссылок. Он может быть скопирован, а после удаления последней копии динамически выделенная память автоматически будет освобождена.
В конце будет предоставлен 1 файл safe_ptr.h, который достаточно подключить через #include “safe_ptr.h”, чтобы вы смогли использовать данный класс.

Базовые принципы многопоточного обмена данными

Как известно, мы можем читать и изменять один и тот же объект из разных потоков только в следующих 4-х случаях:
1. Lock-based. Объект защищен блокировкой: spin-lock, std (mutex, recursive_mutex, timed_mutex, shared_timed_mutex, shared_mutex…): en.cppreference.com/mwiki/index.php?title=Special%3ASearch&search=mutex
2. Atomic. Объект имеет тип std::atomic<T>, где T – указатель, bool или интегральный тип (std::is_integral<T>::value == true), и только если для типа T существуют атомарные операции на уровне CPU: en.cppreference.com/w/cpp/atomic/atomic
2+1 (Lock-based-Atomic) Иначе, если тип T является trivially copyable type, т.е. удовлетворяют условию std::is_trivially_copyable<T>::value == true, тогда std::atomic<T> работает как Lock-based – внутри него автоматически используется блокировка
3. Transaction-safe. Функции, реализованные для работы с объектом, обеспечивают потоко-безопасную гарантию transaction_safe (Transactional Memory TS (ISO/IEC TS 19841:2015) — Experimental C++): en.cppreference.com/w/cpp/language/transactional_memory
4. Lock-free. Функции для работы с объектом реализованы на основе lock-free алгоритмов, т.е. обеспечивают потоко-безопасную гарантию lock-free

Если вы отлично знаете все 4 способа обеспечения потоко-безопасности, то можете пропустить эту главу.
Рассмотрим в обратном порядке:

(4) Lock-free алгоритмы достаточно сложные, и часто для создания каждого сложного алгоритма требуется несколько научных работ. Примеры lock-free алгоритмов для работы с контейнерами: unordered_map, ordered_map, queue… и ссылки на научные работы можно найти в библиотеке – Concurrent Data Structures (CDS) library: github.com/khizmax/libcds
Это очень быстрые и надежные многопоточные структуры данных, но пока что нет планов включать их в стандартную библиотеку C++17 или C++20 и они не включены в черновик стандартов: www.open-std.org/JTC1/SC22/WG21

(3) Transaction-safe планируется включить в экспериментальную часть стандарта C++ и уже есть черновик стандарта ISO/IEC TS 19841:2015: www.open-std.org/jtc1/sc22/wg21/docs/papers/2015/n4514.pdf
Но даже не все STL-контейнеры планируется сделать transaction-safe. Например, контейнер std::map даже не планируется делать транзакционно-безопасным, т.к. для него определены как transaction-safe только следующие функции: begin, end, size, max_size, empty. Но не определены как потоко-безопасные функции: find, modify, insert. А реализовать собственный объект с transaction-safe функциями-членами совсем не легко, иначе это бы сделали для std::map.

(2) Atomic. Этот подход уже стандартизован в C++11 и вы можете легко его использовать. Например, достаточно объявить переменную std::atomic shared_val; затем передать ссылку или указатель на неё в несколько потоков и все обращения через функции-члены и операторы std::atomic будут потоко-безопасны: [3] coliru.stacked-crooked.com/a/9d0219a5cfaa3cd5
Member functions, Specialized member functions: en.cppreference.com/w/cpp/atomic/atomic
Важно понимать, что если вы делаете несколько операций над атомарной переменной, неважно в одном выражении или в нескольких, то между ними другой поток может изменить значение это переменной. Поэтому для атомарного выполнения нескольких операций используется Lock-free подход на основе CAS-функции (Compare-And-Swap) compare_exchange_weak() – а именно: считываем значение из атомарной переменной в локальную переменную (old_local_val), делаем множество нужных нам операций и результат записываем в локальную переменную (new_local_val), и в конце CAS-функцией сравниваем текущее значение атомарной переменной с начальным (old_local_val) и если они не равны, то выполняем цикл заново, а если равны – значит за это время другой поток не вносил изменений, тогда меняем значение атомарной переменной на новое значение (new_local_val). Причем сравнение и присвоение делаем одной операцией compare_exchange_weak() – это атомарная функция и пока она полностью не выполнится, никто не может изменить значение переменной: [4] coliru.stacked-crooked.com/a/aa52b45150e5eb0a
Этот подход с циклом известен под названием – оптимистическая блокировка. Пессимистическими блокировками называются: spin-lock, mutex…
А если все операции такого цикла выполняются без пессимистических блокировок, то такой алгоритм называется lock-free.
Часто атомарной CAS-функцией подменяют указатели, а именно: выделяется новая память, в неё копируется изменяемый объект и получается указатель на него, выполняется ряд действий над этой копией, и в конце CAS-функцией заменяется старый указатель на указатель нового объекта, если за это время другой поток не изменил старый указатель. Но если указатель был изменен другим потоком, то все повторяется сначала.
Здесь возможна проблема называемая «ABA»: en.wikipedia.org/wiki/ABA_problem Когда другие потоки успевают дважды изменить указатель, причем второй раз указатель изменяется на исходное значение, но по этому адресу другие потоки уже успевают удалить объект и создать новый. Т.е. тоже значение указателя указывает уже на другой объект, а мы видим, что значение не изменилось и думаем, что объект не заменялся. Есть множество способов решить эту проблему, например: LL/SC, RCU, hazard_pointer, garbage collector…
Atomic — это самый быстрый способ обмена данными между потоками. Кроме того, для всех атомарных операций можно использовать менее строгие и более быстрые барьеры памяти, которые мы очень подробно рассмотрим далее. По умолчанию же используется самый безопасный и строгий барьер переупорядочивания: std::memory_order_seq_cst. Но как мы заметили выше: нужно много усилий, чтобы реализовать сложную логику используя атомарные переменные.

(2) – (1) Atomic & Lock-based.
Но если вам необходимо прочитать или изменить атомарно сразу несколько переменных std::atomic a, b, c; и вы не хотите реализовывать lock-free алгоритм и решать ABA-проблему, то необходимо использовать блокировку. Процессорная атомарная CAS-функция на большинстве CPU может проверить была ли изменена только одна переменная шириной максимум 64-bit, но в это время могла измениться уже другая переменная. Решение: std::atomic<T> позволяет использовать для T – тип структуры любого размера.
В стандарте C++ введена возможность использования любого типа T для std::atomic<T>, если он «trivially copyable type», т.е. удовлетворяет условию std::is_trivially_copyable<T>::value == true
Что говорит стандарт C++ Working Draft, Standard for Programming Language C++ N4606 2016-07-12: www.open-std.org/jtc1/sc22/wg21/docs/papers/2016/n4606.pdf

§29.5 / 1
There is a generic class template atomic<T>. The type of the template argument T shall be trivially copyable (3.9). [ Note: Type arguments that are not also statically initializable may be difficult to use. — end note ]

§3.9 / 9
Scalar types, trivially copyable class types (Clause 9), arrays of such types, and cv-qualified versions of these types (3.9.3) are collectively called trivially copyable types

Но если процессорная атомарная CAS-функция может проверить была ли изменена только одна переменная шириной максимум 64-bit, а у нас 3 переменные по 32-bit, то как же сработает CAS-функция в std::atomic<T>? CAS и все остальные функции будут автоматически использовать блокировку (std::mutex или какую-то другую), которая содержится в стандартной реализации std::atomic<T>, для T – trivially copyable types.
Для атомарного изменения нескольких переменных – мы можем использовать структуру из переменных struct T { int price, count, total; }; как тип для шаблона std::atomic<T>.
Пример: [5] coliru.stacked-crooked.com/a/bdfb3dc440eb6e51

Example output: 10, 7, 70

В этом примере последнее значение 70 в любой момент времени будет равно произведению первых двух значений 10*7 – т.е. вся структура меняется только атомарно.
Этот код в gcc и clang для x86 необходимо компилировать с флагом -latomic
При этом каждое обращение к std::atomic<T> shared_val; будет вызывать блокировку внутри неё, о чем говорит значение функции shared_val.is_lock_free() == false.
Т.е. глобально мы использовали оптимистическую блокировку (цикл), а локально использовали 2 пессимистические блокировки при обращении к атомарной переменной: получении старого значения и вызове CAS-функции.

(1) Lock-based.
Но мы не сможем использовать std::atomic<T> для любых типов T созданных вами из-за обязательного условия «trivially copyable» для типа T. Из всех STL-контейнеров мы можем использовать только std::array<>. Например, не можем использовать std::atomic<std::map<int, int>>, т.к. тип std::map<> не trivially copyable, для любых аргументов его шаблона. И ваши собственные классы скорее всего тоже не смогут быть использованы в качестве типа T для std::atomic<T>.
В этом случае вам придется самим создавать объекты мьютексов, блокировать их каждый раз перед использованием разделяемых (shared) объектов и разблокировать после. Concept: en.cppreference.com/w/cpp/concept/Mutex
В C++ имеются следующие мьютексы: std::mutex, std::recursive_mutex, std::timed_mutex, std::recursive_timed_mutex, std::shared_timed_mutex, std::shared_mutex. Подробнее о них написано здесь: en.cppreference.com/w/cpp/thread
Например, создаем любой разделяемый между потоками объект std::map<int, T> и создаем мьютекс защищающий его, затем передаем ссылки на них в несколько потоков. И в каждом потоке блокируем мьютекс перед использованием разделяемого объекта: [6] coliru.stacked-crooked.com/a/a97e905d54ae1fbb
Причем блокируем с использованием RAII-идиомы: std::lock_guard<std::mutex> lock(mtx); – при создании этого объекта его конструктор блокирует мьютекс, а в конце жизни объекта деструктор разблокирует мьютекс. Таким образом мы точно не забудем его разблокировать, т.к. деструктор вызовется автоматически.
Но остаются ещё 4 основные проблемы:

1. Взаимоблокировки (deadlocks) – если вы напишите код так, что поток-1 заблокирует mtx1, а поток-2 заблокирует mtx2, и удерживая блокировку поток-1 попробуют захватить mtx2, а поток-2 попробует захватить mtx1, то эти потоки будут ждать друг друга вечно. Этой проблемы лишены lock-free алгоритмы, но не любую логику можно реализовать с помощью lock-free – это мы покажем на примере согласованного атомарного изменения нескольких контейнеров.
2. Если вы напишите код так, что пока мьютекс заблокирован вы присвоите ссылку на разделяемый объект указателю, жизнь которого дольше, чем у блокировки std::lock_guard, то после снятия блокировки вы сможете по этому указателю обратиться к разделяемому объекту – это приведет к проблеме Data races, т.е. к не консистентному состоянию разделяемого объекта и к неверной работе или к краху программы. Тоже самое будет если итератор полученный от разделяемого объекта будет использоваться после разблокировки мьютекса.
3. Вы можете перепутать мьютексы и заблокировать мьютекс, который защищает другой объект – Data races.
4. Вы можете просто забыть заблокировать мьютекс в нужном месте – Data races.

Execute Around Pointer Idiom

Помимо RAII-идиомы есть ещё одна интересная идиома – Execute Around Pointer, которая поможет справиться с последними двумя проблемами:

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

В результате вы не сможете забыть заблокировать мьютекс, и не сможете перепутать мьютексы.

Делаем любой объект потоко-безопасным

Execute Around Pointer Idiom – это давно известная идиома со строго определенным порядком исполнения, применяемая в различных целях от визуализации до логирования: en.wikibooks.org/wiki/More_C%2B%2B_Idioms/Execute-Around_Pointer
Пример: [7] coliru.stacked-crooked.com/a/4f3255b5473c73cc

  execute_around<std::vector<int>> vecc(10, 10); 
  int res = my_accumulate(vecc->begin(), vecc->end(), 0);

Сначала будут созданы временные объекты типа proxy, которые заблокируют мьютексы внутри execute_around, затем в функцию будут переданы итераторы возвращенные функциями begin() и end(), затем будет исполнена функция my_accumulate(), и только после её завершения временные объекты с типом proxy будут удалены, а их деструкторы разблокируют мьютексы.
Подробнее можно прочитать в статье: C++ Patterns Executing Around Sequences. Kevlin Henney: hillside.net/europlop/HillsideEurope/Papers/ExecutingAroundSequences.pdf
В C++ есть два определения, которые строго определяют порядок выполнения операций Standard § 1.9 (13): sequenced before и sequenced after. В приведенных ниже цитатах стандарта вы 2 раза увидите «sequenced before».
Принцип и последовательность выполнения всех действий в Execute Around Pointer Idiom строго описаны в стандарте C++ Working Draft, Standard for Programming Language C++ N4606 2016-07-12: www.open-std.org/jtc1/sc22/wg21/docs/papers/2016/n4606.pdf
Сначала приведем пять цитат из стандарта, а затем покажем, как каждая из цитат объясняет поведение Execute Around Pointer Idiom.

1. Для всех типов T отличных от raw-pointers: x->m интерпретируется как (x.operator->())->m. Т.е. выражение (x->m) будет многократно разворачиваться ((x.operator->().operator->())->m, пока не получим raw-указатель. Пример с тремя идентичными по смыслу выражениями: [8] coliru.stacked-crooked.com/a/dda2474024b78a0b

§ 13.5.6
operator-> shall be a non-static member function taking no parameters. It implements the class member access syntax that uses ->.
postfix-expression -> template opt id-expression
postfix-expression -> pseudo-destructor-name
An expression x->m is interpreted as (x.operator->())->m for a class object x of type T if T::operator->() exists and if the operator is selected as the best match function by the overload resolution mechanism (13.3).

2. Когда функция вызывается, даже если она «inline», абсолютно любые вычисления и эффекты от выражений, вычисляющих аргументы функции – исполняются до того, как тело функции начнет исполняться.

§ 1.9 / 16
When calling a function (whether or not the function is inline), every value computation and side effect associated with any argument expression, or with the postfix expression designating the called function, is sequenced before execution of every expression or statement in the body of the called function.

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

§ 1.9 / 10

void f() {
if (S(3).v()) 	// full-expression includes lvalue-to-rvalue and
// int to bool conversions, performed before
// temporary is deleted at end of full-expression
{ }
}

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

§ 1.9 Footnote 8
As specified in 12.2, after a full-expression is evaluated, a sequence of zero or more invocations of destructor functions for temporary objects takes place, usually in reverse order of the construction of each temporary object.

5. Три случая, когда временный объект уничтожается не в конце всего выражения – 2 случая при инициализации элементов массива, а 3-й случай, когда создана ссылка на временный объект.

§ 12.2 Temporary objects
§ 12.2 / 5
There are three contexts in which temporaries are destroyed at a different point than the end of the full expression. The first context is when a default constructor is called to initialize an element of an array with no corresponding initializer (8.6). The second context is when a copy constructor is called to copy an element of an array while the entire array is copied (5.1.5, 12.8). In either case, if the constructor has one or more default arguments, the destruction of every temporary created in a default argument is sequenced before the construction of the next array element, if any. The third context is when a reference is bound to a temporary.

Например, у нас есть упрощенный класс execute_around<>

template<typename T, typename mutex_type = std::recursive_mutex>
class execute_around {
  std::shared_ptr<mutex_type> mtx;
  std::shared_ptr<T> p;
 
  void lock() const { mtx->lock(); }
  void unlock() const { mtx->unlock(); }
  public:
    class proxy {
      std::unique_lock<mutex_type> lock;
      T *const p;
      public:
        proxy (T * const _p, mutex_type& _mtx) : lock(_mtx), p(_p)  { std::cout << "locked n";} 
        proxy(proxy &&px) : lock(std::move(px.lock)), p(px.p)  {}
        ~proxy () { std::cout << "unlocked n"; }
        T* operator -> () {return p;}
        const T* operator -> () const {return p;}
    };
 
    template<typename ...Args>
    execute_around (Args ... args) : 
        mtx(std::make_shared<mutex_type>()), p(std::make_shared<T>(args...))  {}  
 
    proxy operator -> () { return proxy(p.get(), *mtx); }
    const proxy operator -> () const { return proxy(p.get(), *mtx); }
    template<class Args> friend class std::lock_guard;
};

И мы используем наш шаблонный класс execute_around<> следующим образом, пример:[45] coliru.stacked-crooked.com/a/d2e02b61af6459f5

int main() {
    typedef execute_around<std::vector<int>> T;
    T vecc(10, 10); 
    int res = my_accumulate(vecc->begin(), vecc->end(), 0);
    return 0;
}

Тогда последнее выражение можно через несколько преобразований привести к следующему виду.
1. Согласно 1-й цитате стандарта, x->m интерпретируется как (x.operator->())->m:

  int res = my_accumulate(
(vecc.operator->())->begin(),
(vecc.operator->())->end(), 
0);

2. Т.к. vecc.operator->() возвращает временный объект T::proxy(), получаем:

  int res = my_accumulate( 
T::proxy(vecc.p.get(), *vecc.mtx)->begin(),
T::proxy(vecc.p.get(), *vecc.mtx)->end(),
0);

3. Далее согласно цитатам 2, 3 и 4 – временные объекты типа proxy будут созданы до того, как функция начнет выполняться, и будут уничтожены после окончания функции (после окончания всего выражения):

T::proxy tmp1(vecc.p.get(), *vecc.mtx);	// lock-1 std::recursive_mutex
T::proxy tmp2(vecc.p.get(), *vecc.mtx);	// lock-2 std::recursive_mutex

  int res = my_accumulate(tmp1->begin(), tmp2->end(), 0);

tmp2.~ T::proxy();				// unlock-2 std::recursive_mutex
tmp1.~ T::proxy();				// unlock-1 std::recursive_mutex

4. Согласно снова 1-й цитате:
• tmp1->begin() эквивалентно (tmp1.operator->())->begin()
• tmp1.operator->() возвращает p
В итоге получаем, где p – это shared_ptr-указатель на наш объект типа std::vector:

typedef execute_around<std::vector<int>> T;
    T vecc(10, 10); 
T::proxy tmp1(vecc.p.get(), *vecc.mtx);	// lock-1 std::recursive_mutex
T::proxy tmp2(vecc.p.get(), *vecc.mtx);	// lock-2 std::recursive_mutex

  int res = my_accumulate(tmp1.p->begin(), tmp2.p ->end(), 0);

tmp2.~ T::proxy();				// unlock-2 std::recursive_mutex
tmp1.~ T::proxy();				// unlock-1 std::recursive_mutex

За 4 шага мы описали строгую последовательность всех действий идиомы. Заметим, что стандарт не гарантирует взаимный порядок создания временных переменных tmp1 и tmp2, т.е. сначала может быть создан tmp2, а затем tmp1, но это не меняет логику нашей программы.
Заметим, что мы не обратились к 5-ой цитате стандарта, т.к. в ней описаны 3 случая, когда время удаления объекта может отличаться от приведенного, и как мы видим ни один из этих случаев не может соответствовать нашему. Первые 2 случая в цитате стандарта – это инициализация или копирование массива, они укорачивают жизнь временного объекта, а 3-й случай – это продление жизни временно объекта за счет наличия на него ссылки.

Потоко-безопасный ассоциативный массив

Согласитесь, было бы удобно иметь такой шаблонный класс safe_ptr<> в который можно передать любой тип, и как результат получить потоко-безопасный результирующий тип?
safe_ptr<std::map<std::string, std::pair<std::string, int> >> safe_map_strings;
Далее с этим объектом можно работать, как с указателем на ассоциативный массив: std::shared_ptr<std::map<std::string, std::pair<std::string, int> >>
Но теперь мы можем безопасно работать с ним из разных потоков, и каждое отельное выражение будет потоко-безопасным:

    (*safe_map_strings)["apple"].first = "fruit";
    (*safe_map_strings)["potato"].first = "vegetable";

    safe_map_strings->at("apple").second = safe_map_strings->at("apple").second * 2;
    safe_map_strings->find("potato")->second.second++;

Приведем полностью рабочий пример потоко-безопасного ассоциативного: std::map<>.[9]
coliru.stacked-crooked.com/a/5def728917274b22

#include <iostream>
#include <string>
#include <vector>
#include <memory>
#include <mutex>
#include <thread>
#include <map>

template<typename T, typename mutex_t = std::recursive_mutex, typename x_lock_t = 
std::unique_lock<mutex_t>, typename s_lock_t = std::unique_lock<mutex_t >>
class safe_ptr {
    typedef mutex_t mtx_t;
    const std::shared_ptr<T> ptr;
    std::shared_ptr<mutex_t> mtx_ptr;

    template<typename req_lock>
    class auto_lock_t {
        T * const ptr;
        req_lock lock;
    public:
        auto_lock_t(auto_lock_t&& o) : ptr(std::move(o.ptr)), lock(std::move(o.lock)) { }
        auto_lock_t(T * const _ptr, mutex_t& _mtx) : ptr(_ptr), lock(_mtx){}
        T* operator -> () { return ptr; }
        const T* operator -> () const { return ptr; }
    };

    template<typename req_lock>
    class auto_lock_obj_t {
        T * const ptr;
        req_lock lock;
    public:
        auto_lock_obj_t(auto_lock_obj_t&& o) : 
ptr(std::move(o.ptr)), lock(std::move(o.lock)) { }
        auto_lock_obj_t(T * const _ptr, mutex_t& _mtx) : ptr(_ptr), lock(_mtx){}
        template<typename arg_t>
        auto operator [] (arg_t arg) -> decltype((*ptr)[arg]) { return (*ptr)[arg]; }
    };

    void lock() { mtx_ptr->lock(); }
    void unlock() { mtx_ptr->unlock(); }
    friend struct link_safe_ptrs;
    template<typename mutex_type> friend class std::lock_guard;
    //template<class... mutex_types> friend class std::lock_guard;    // C++17
public:
    template<typename... Args>
    safe_ptr(Args... args) : ptr(std::make_shared<T>(args...)), mtx_ptr(std::make_shared<mutex_t>()) {}

    auto_lock_t<x_lock_t> operator -> () { return auto_lock_t<x_lock_t>(ptr.get(), *mtx_ptr); }
    auto_lock_obj_t<x_lock_t> operator * () { return auto_lock_obj_t<x_lock_t>(ptr.get(), *mtx_ptr); }
    const auto_lock_t<s_lock_t> operator -> () const { return auto_lock_t<s_lock_t>(ptr.get(), *mtx_ptr); }
    const auto_lock_obj_t<s_lock_t> operator * () const { return auto_lock_obj_t<s_lock_t>(ptr.get(), *mtx_ptr); }
};
// ---------------------------------------------------------------


safe_ptr<std::map<std::string, std::pair<std::string, int> >> safe_map_strings_global;


void func(decltype(safe_map_strings_global) safe_map_strings) 
{
    //std::lock_guard<decltype(safe_map_strings)> lock(safe_map_strings);

    (*safe_map_strings)["apple"].first = "fruit";
    (*safe_map_strings)["potato"].first = "vegetable";

    for (size_t i = 0; i < 10000; ++i) {
        safe_map_strings->at("apple").second++;
        safe_map_strings->find("potato")->second.second++;
    }

    auto const readonly_safe_map_string = safe_map_strings;

    std::cout << "potato is " << readonly_safe_map_string->at("potato").first <<
        " " << readonly_safe_map_string->at("potato").second <<
        ", apple is " << readonly_safe_map_string->at("apple").first <<
        " " << readonly_safe_map_string->at("apple").second << std::endl;
}


int main() {

    std::vector<std::thread> vec_thread(10);
    for (auto &i : vec_thread) i = std::move(std::thread(func, safe_map_strings_global)); 
    for (auto &i : vec_thread) i.join();

    std::cout << "end";
    int b; std::cin >> b;

    return 0;
}

Вывод:
potato is vegetable 65042, apple is fruit 65043
potato is vegetable 81762, apple is fruit 81767
potato is vegetable 84716, apple is fruit 84720
potato is vegetable 86645, apple is fruit 86650
potato is vegetable 90288, apple is fruit 90291
potato is vegetable 93070, apple is fruit 93071
potato is vegetable 93810, apple is fruit 93811
potato is vegetable 95788, apple is fruit 95790
potato is vegetable 98951, apple is fruit 98952
potato is vegetable 100000, apple is fruit 100000
end

Отсюда 2 вывода:

1. Итоговые значения равные 100000 означают, что каждое сложение в каждом из 10 потоков было выполнено потоко-безопасно. Действительно, достаточно изменить наш код, чтобы operator -> возвращал указатель на сам объект вместо типов auto_lock_t и auto_lock_obj_t , как мы увидим, что было бы если бы код НЕ был потоко-безопасным – data-race: [10] coliru.stacked-crooked.com/a/45d47bcb066adf2e
2. Промежуточные значения не кратные 10000 говорят о том, что потоки исполнялись параллельно или псевдопараллельно, т.е. прерывались посреди любой из операции и в это время исполнялся другой поток. Т.е. перед каждым инкрементом operator++ блокировался мьютекс, и сразу после инкремента разблокировался, а далее мьютекс мог блокироваться другим потоком, который выполнял инкремент. Мы можем вначале каждого потока сразу заблокировать мьютекс до конца исполнения функции потока используя std::lock_guard<>, и увидим какой-бы был результат, если бы потоки исполнялись последовательно, а не псевдопараллельно: [11] coliru.stacked-crooked.com/a/cc252270fa9f7a78

Оба вывода подтверждают, что наш шаблон класса умного указателя safe_ptr<T> автоматически обеспечивает потоко-безопасность защищаемого объекта типа T.

Потоко-безопасность, атомарность и согласованность нескольких объектов.

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

1. user_accounts(INT user_id, STRING user_name, INT money) – таблица с количеством денег для каждого клиента – отсортирована по полю user_id
2. cash_flows(INT unique_id, INT src_id, INT dst_id, INT time, INT money) – таблица показывающая движение денег – на каждую запись ссылаются два ассоциативных массива, отсортированные: по полю src_id и по полю dst_id

// 1-st table 
struct user_accounts_t { 
    std::string user_name; int64_t money; 
    user_accounts_t(std::string u, int64_t m) : user_name(u), money(m) {}
};

std::map<uint64_t, user_accounts_t> user_accounts;

// 2-nd table 
struct cash_flows_t { uint64_t unique_id, src_id, dst_id, time; int64_t money; };

std::atomic<uint64_t> global_unique_id;	// SQL-sequence
std::multimap<uint64_t, std::shared_ptr<cash_flows_t>> cash_flows_src_id;
std::multimap<uint64_t, std::shared_ptr<cash_flows_t>> cash_flows_dst_id;

В терминах СУБД (RDBMS):

• 1-я таблица с индексом по полю user_id – это Index-Organized-Table (Oracle) или Table with Clustered Index (MS SQL).
• 2-я таблица – это таблица с двумя упорядоченными индексами по одному полю src_id и одному полю dst_id соответственно.

В реальных задачах в таблице может находиться миллионы записей для клиентов и миллиарды записей для денежных движений, в этом случае индексы по полям user_id, src_id, dst_id ускоряют поиск по ним в сотни тысяч раз, поэтому они крайне необходимы.
Предположим от трех пользователей пришли запросы на выполнение трех задач в трех параллельных потоках:
1. move_money() – поток переводит деньги от одного клиента другому

• a) у одного клиента отнимает деньги
• b) другому клиенту прибавляет эту же сумму денег
• c) в индекс по полю id-source добавляет указатель на денежную проводку
• d) в индекс по полю id-destination добавляет указатель на эту же денежную проводку (в реальных задачах это не обязательно, но для примера мы это сделаем)

2. show_total_amount() – показать сумму денег всех клиентов

• a) В цикле проходим по каждому клиенту и суммируем все деньги

3. show_user_money_on_time() – показать количество денег клиента с заданными user_id на момент времени time

• a) incoming – суммируем все деньги, которые пришли клиенту с момента time и позднее (используя индекс по полю id-source)
• b) outcoming – суммируем все деньги, которые ушли от клиента с момента time и позднее (используя индекс по полю id-destination)
• c) user_money – получим текущие деньги у клиента
• d) user_ user_money — incoming + outcoming – это количество денег клиента на момент времени time

Мы знаем, что любой из потоков может быть прерван в любой момент операционной системой, например, чтобы отдать CPU-Core другому потоку. Самое опасное, что это происходит крайне редко и вы можете ни разу не встретиться с этим при отладке, но однажды это произойдет у клиента, и если это приведет к data-races, то в финансовой системе могут просто исчезнуть деньги.
Поэтому мы преднамеренно добавим функции ожидания, которые уложат спать поток на несколько миллисекунд в наиболее критичных местах, чтобы сразу увидеть ошибки.
Мы сделаем потоко-безопасными наши таблицы user_accounts, cash_flows_src_id, cash_flows_dst_id используя safe_ptr<>, но станет ли от этого потоко-безопасной вся программа?
[12] coliru.stacked-crooked.com/a/5bbba59b63384a2b
Посмотрим на «основные строки» в выводе программы, помеченные <<<:

Init table safe_user_accounts:
at time = 0 <<<
1 => John Smith, 100
2 => John Rambo, 150
— start transaction… show_total_amount()
1 => John Smith, 100
2 => John Rambo, 100
Result: all accounts total_amount = 200 <<<

— start transaction… show_user_money_on_time()
1 => John Smith, 150, at time = 0 <<<

Сразу видны две ошибки:

1. Изначально в сумме у всех (двух) пользователей было 250 денег, а функция show_total_amount() показала только 200, ещё 50 куда-то пропали
2. На момент времени time=0 у пользователя 1 было 100 денег, а функция show_user_money_on_time() показала неверный результат – у пользователя 1 было 150 денег на момент time=0

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

Правильный пример: [13] coliru.stacked-crooked.com/a/c8bfc7f5372dfd0c
Посмотрим на «основные строки» в выводе программы, помеченные <<<:

Init table safe_user_accounts:
at time = 0 <<<
1 => John Smith, 100
2 => John Rambo, 150

Result: all accounts total_amount = 250 <<<

1 => John Smith, 100, at time = 0 <<<

Теперь все верно, сумма денег всех клиентов 250, а количество денег у клиента 1 была 100 на момент времени 0.
Т.е. мы смогли атомарно выполнить операции не с одним объектом, а сразу с 3-мя объектами, сохраняя согласованность данных для любых операций.
Но здесь возможна другая проблема. Если вы или другой разработчик в какой-то из функций заблокирует мьютексы контейнеров в другом порядке, то возможна ситуация под названием deadlock – когда 2 потока повиснут вечно ожидая друг друга.
В правильном примере выше мы блокировали мьютексы в обоих функциях move_money() и show_user_money_on_time() в одинаковом порядке:

• lock1(safe_user_accounts)
• lock2(safe_cash_flows_src_id)
• lock3(safe_cash_flows_dst_id)

А теперь посмотрим, что будет если мы заблокируем мьютексы в контейнерах в каждой функции в разном порядке:
• move_money()

• lock2(safe_cash_flows_src_id)
• lock3(safe_cash_flows_dst_id)
• lock1(safe_user_accounts)

• show_user_money_on_time()

• lock3(safe_cash_flows_dst_id)
• lock2(safe_cash_flows_src_id)
• lock1(safe_user_accounts)

Функция move_money() заблокировала lock2 и ждет пока освободиться lock3, чтобы его заблокировать. Функция show_user_money_on_time() заблокировала lock3 и ждет пока освободиться lock2, чтобы его заблокировать. И ждать они друг друга будут вечно.
Пример: [14] coliru.stacked-crooked.com/a/0ddcd1ebe2be410b
Вывод:

Init table safe_user_accounts:
at time = 0 <<<
1 => John Smith, 100
2 => John Rambo, 150

— start transaction… move_money()
— start transaction… show_total_amount()

1 => John Smith, 100
2 => John Rambo, 150

Т.е. функции move_money() и show_user_money_on_time() не были завершены и остановились навечно в deadlock.
Решения есть 4:
1. Все разработчики во всех функциях всегда блокируют мьютексы в одном и том же порядке и никогда не ошибаются – это очень ненадежное предположение
2. Вы изначально объединяете в одну структуру все объекты, которые будут использоваться атомарно, и используете безопасный указатель с типом этой структуры struct all_t { std::map<int,int> m1; std::multimap<int,int> m2; … }; safe_ptr<all_t> safe_all_obj; – но если вы изначально использовали эти 2 контейнера только отдельно safe_ptr<map<int,int>> m1; safe_ptr<multimap<int,int>> m2; и уже написали много кода, а затем решили их объединить в одну структуру защищенную одним мьютексом, то вам придется переписать все места где их используете, например, вместо m2->at(5); необходимо safe_all_obj->m2.at(5); Переписывать много кода – это не очень удобно.
3. Вы можете один раз объединить safe_ptr<> которые используются совместно, чтобы они использовали один и тот же рекурсивный мьютекс, после чего будет неважно в каком порядке они блокируются, всегда будет сохраняться консистенность этих объектов и никогда не будет deadlock. Для этого необходимо лишь добавить 1 строчку – это очень удобно. Но это может снизить производительность, т.к. теперь блокировка одного из контейнеров всегда приводит к блокировке всех связанных с ним контейнеров. Вы получите консистентность даже тогда, когда она не нужна – ценой снижения производительности. Пример: [15] coliru.stacked-crooked.com/a/2a6f1535e0b95f7b
Все изменения в коде – это всего лишь одна строчка:
static link_safe_ptrs tmp_link(safe_user_accounts, safe_cash_flows_src_id, safe_cash_flows_dst_id);
Вывод – показаны основные строки:

Init table safe_user_accounts:
at time = 0 <<<
1 => John Smith, 100
2 => John Rambo, 150

Result: all accounts total_amount = 250 <<<

1 => John Smith, 100, at time = 0 <<<

4. Вы можете использовать блокировку сразу для нескольких мьютексов разного типа с установкой времени timeout для блокировки каждого мьютекса. А если не удается заблокировать хотя бы один из мьютексов за это время, то все ранее заблокированные мьютексы разблокируются, поток ждет какое-то время и пытается по очереди заблокировать все мьютексы снова. Для этого достаточно добавить по одной строчке перед каждым использованием контейнеров lock_timed_any_infinity lock_all(safe_cash_flows_src_id, safe_cash_flows_dst_id, safe_user_accounts); и не важно в каком порядке вы будете блокировать мьютексы контейнеров. Пример: [16] coliru.stacked-crooked.com/a/93206f216ba81dd6
Т.е. даже если мы блокируем мьютексы в разных последовательностях:

• То при использовании lock_timed_any_infinity не возникает deadlock: [17] coliru.stacked-crooked.com/a/7ac7640918228090
• А в аналогичном примере при использовании lock_guard<> – возникает deadlock: [18] coliru.stacked-crooked.com/a/a281f90299771434

Таким образом мы с помощью блокировок (locks) решили задачу компонуемости (Composable) гарантированно без вечных deadlock: https://en.wikipedia.org/wiki/Lock_(computer_science)#Lack_of_composability

Компонуемость (Composable) и взаимоблокировки (Deadlocks)

Т.к. выше для потоко-безопасности мы использовали блокировки, то наши алгоритмы называются — lock-based.
А действительно ли все хорошо с отсутствием deadlocks у lock-free контейнеров, алгоритмов на основе Transactional Memory, и бывают ли deadlocks в современных СУБД: MSSQL (lock-based IL) и Oracle (multi-version concurrency control).
Алгоритмы lock-free не позволяют изменять атомарно сразу несколько контейнеров. RDBMS (РСУБД) имеют все те же проблемы с deadlock, как в lock-based алгоритмах, и часто решают их так же через таймауты блокировок или графы блокировок. А новый раздел transaction-safe в стандарте C++ не позволяет потко-безопасно использовать сложные алгоритмы такие как std::map<>.
Lock-free алгоритмы не обладают свойством Composable operations – совместного атомарного использования нескольких lock-free алгоритмов. Т.е. не могут быть изменены или прочитаны атомарно сразу несколько lock-free структур данных. Например, вы можете использовать lock-free контейнеры ассоциативных массивов из libCDS, и они по отдельности будут потоко-безопасны. Но если вы захотите атомарно выполнить операции сразу с несколькими lock-free контейнерами и сохранить консистентность, то вы не сможете этого сделать, т.к. их API не предоставляет функции lock-free операций одновременно над несколькими контейнерами. Пока вы меняете или читаете один контейнер, в это время уже будет изменен другой. Чтобы этого избежать – вам придется использовать блокировки, в этом случае это уже будут контейнеры, основанные на блокировках, а значит им станут свойственны все проблемы lock-based алгоритмов, а именно возможность проблемы взаимоблокировок (deadlocks). Кроме того, блокировки иногда используются и при использовании всего одного контейнера:

• libcds.sourceforge.net/doc/cds-api/namespacecds_1_1sync.html
• github.com/khizmax/libcds/tree/master/cds/lock
• github.com/khizmax/libcds/tree/master/cds/sync

В транзакционных RDBMS таких как MSSQL (lock-based) и Oracle (multi-version concurrency control) так же используются блокировки и именно поэтому есть проблемы deadlocks, которые, например, автоматически могут решаться через построение графа блокировок и нахождение циклических ожиданий, или через установку времени ожидания блокировки select col from tbl where id in (....) for update wait 30; Если время ожидания истекло или найден deadlock в графе блокировок, то происходит rollback (откат) одной из транзакции – т.е. отмена всех изменений, которые уже были внесены этой транзакцией, разблокировка всего что было заблокировано, а затем вы можете попытаться выполнить транзакцию с самого начала (и так много раз):

• Oracle Database Online Documentation 11g Release 1 (11.1) — Deadlocks: docs.oracle.com/cd/B28359_01/server.111/b28318/consist.htm#i5337
  Блокировки Oracle: docs.oracle.com/cd/B28359_01/server.111/b28318/consist.htm#i5249
  Как видно, эксклюзивные несовместимые блокировки используются любыми выражениями: Insert/Update/Delete/Select-For-Update
• MSSQL Detecting and Ending Deadlocks: https://technet.microsoft.com/en-us/library/ms178104(v=sql.105).aspx
  Блокировки MS SQL: https://technet.microsoft.com/en-us/library/ms186396(v=sql.105).aspx

В свою очередь Transactional Memory в отличие от Lock-free контейнеров может атомарно работать со множеством контейнеров/данных. Т.е. Transactional Memory обладает свойством Composable operations. Внутри используются или пессимистические блокировки (с вероятностью deadlock conflict) или оптимистические блокировки (с большей вероятностью conflicting modifications при конкурентных модификациях). А при любых конфликтах транзакция автоматически отменяется и повторяется с самого начала, что влечет многократное повторение всех операций – а это несёт большие накладные расходы. Эти накладные расходы пытаются уменьшить за счет создания Hardware Transactional Memory на уровне CPU, но пока что нет реализаций показывающих приемлемую производительность, хотя Intel уже добавила Hardware Transactional Memory в CPU Haswell. В стандарте C++ так же обещают включить Transactional Memory, но только в будущем, пока что только как экспериментальную и без поддержки работы с std::map. Т.е. пока что красиво все только в теории. Но в будущем это скорее всего заменит привычные нам методы синхронизации.
Итого:

• Lock-based алгоритмы не компонуются, если такая возможность не заложена при их реализации, но такую возможность можно реализовать, и мы успешно её реализовали в предыдущих разделах.
• Lock-free алгоритмы не компонуются и скомпоновать их без блокировок это сверх сложная задача, а с блокировками такой алгоритм уже не являются lock-free и появляется риск возникновения вечных deadlocks.
• RDBMS: MSSQL(lock-based IL) и Oracle (MVCC) – возможны вечные взаимоблокировки (deadlocks), которые снимаются через графы блокировок или таймауты
• Транзакционная память из экспериментальной части стандарта C++ – пока ограничена использованием только в самых простых алгоритмах и не позволяет использовать алгоритмы такие, как в методах std::map<> или сложнее.

Текущий вывод: проблема deadlock есть во всех типах алгоритмов и систем показывающих высокую производительность, где задействуется одновременно несколько структур данных, и мы предложили 2 варианта её решения для safe_ptr<>

1. static link_safe_ptrs tmp_link(safe_user_accounts, safe_cash_flows_src_id, safe_cash_flows_dst_id); — использовать один мьютекс для нескольких контейнеров
2. lock_timed_any_infinity lock_all(safe_cash_flows_src_id, safe_cash_flows_dst_id, safe_user_accounts);- использовать таймауты блокировок, а по окончании времени разблокировать всё и пытаться заблокировать заново

В случае же когда для safe_ptr<> используется только 1 контейнер и рекурсивный мьютекс, то deadlock невозможен в safe_ptr<>, т.к. для deadlock необходимо как минимум 2 рекурсивных мьютекса (либо 1 нерекурсивный).

Компонуемость (Composable) lock-based алгоритмов

В общем случае считается, что lock-based программы не компонуемы (Composable), т.е. если просто взять 2 lock-based структуры данных и атомарно по очереди их изменить, то не получится консистентное состояние в любой момент времени.
Но выше мы легко скомпоновали три lock-based контейнера, как же нам это удалось? Есть небольшое уточнение на этот счет – выделенное жирным шрифтом:

Perhaps the most fundamental objection [...] is that lock-based programs do not compose: correct fragments may fail when combined. For example, consider a hash table with thread-safe insert and delete operations. Now suppose that we want to delete one item A from table t1, and insert it into table t2; but the intermediate state (in which neither table contains the item) must not be visible to other threads. Unless the implementor of the hash table anticipates this need, there is simply no way to satisfy this requirement. [...] In short, operations that are individually correct (insert, delete) cannot be composed into larger correct operations.

—Tim Harris et al., «Composable Memory Transactions», Section 2: Background, pg.2[6]
www.microsoft.com/en-us/research/wp-content/uploads/2005/01/2005-ppopp-composable.pdf
Дело в том, что lock-based алгоритмы не компонуются, если такая возможность не заложена при их реализации. Т.е. lock-based структуры данных не компонуются автоматически, но их можно компоновать вручную, например, как у нас с помощью класса lock_timed_any_infinity, если снаружи есть доступ к их мьютексу для операций компоновки.
Мы реализовали lock-based шаблонный класс safe_ptr<T> и в нем для любого типа T мы предусмотрели необходимость компоноваться и разрешать (solve) deadlocks с использованием операций компоновки: link_safe_ptrs, lock_timed_any_infinity, lock_timed_any_once.

Итак, почему мы выбрали блокировки и их пессимистический вариант?

• Блокировки — это стандартный механизм обеспечения потоко-безопасности операционных систем и языка C++
• С помощью блокировок мы можем реализовать компонуемость и консистентность множества структур данных
• В пессимистических блокировках возможен deadlock если вы забыли правильно скомпоновать блокировки, он трудно находится, но легко решается и возникает редко. В оптимистических блокировках в любом случае возможен conflicting modifications, он легко находится, но требует дополнительного C++-кода для его решения и возникает намного чаще.
• Tom Kyte is a Senior Technical Architect in Oracle’s Server Technology Division – является сторонником пессимистический блокировок в Oracle DB (Multi-Version Concurrency Control): https://asktom.oracle.com/pls/asktom/f?p=100:11:0::::P11_QUESTION_ID:5771117722373
  А о том, что пессимистические блокировки могут вызвать ситуации Lockout и Deadlock он пишет следующее:

  I am a huge fan of so called pessimistic locking. The user has very clearly announced their intent to UPDATE THE DATA. The lockouts they refer to are easily handled with session timeouts (trivial) and the deadlock is so rare and would definitely be an application bug (in both Oracle and RDB).
• Deadlock – это ошибка. Если вы считаете, что лучше небольшое замедление выполнения нескольких потоков вместо их полной остановки пока вы исправляете свою ошибку, то используйте lock_timed_any_infinity. Иначе, если хотите вечную остановку вашей программы, то используйте: link_safe_ptrs и std::lock_guard<>.
• Также нет необходимости в автоматической эскалации блокировок. Например, Oracle DB это никогда не делает: docs.oracle.com/cd/B28359_01/server.111/b28318/consist.htm#CIHCFHGE
  

  Oracle Database never escalates locks. Lock escalation greatly increases the likelihood of deadlocks.

Мы и дальше будем периодически обращаться к богатому опыту промышленных СУБД при реализации наших алгоритмов.
Пример использования safe_ptr<T>
из safe_ptr.h: github.com/AlexeyAB/object_threadsafe/tree/master/any_object_threadsafe

Вывод:

Мы доказали корректность Execute Around Pointer Idiom для автоматического обеспечения безопасного доступа из разных потоков. Показали пример его компонуемости (composable). А так же показали плюсы использования пессимистических блокировок для обеспечения потокобезопасности.

Автор: AlexeyAB

Источник