Наведение порядка в коде

1.

Наведение порядка в коде

2.

Простой пример*
1
2
3
4
5
int count = 0;
#pragma omp parallel for
for (int i = 0; I < n; ++i)
if ( a[i] ==0 )
count ++;
* Не пишите так никогда!!!

3.

Простой пример
Значение count
Поток 1
0
mov ax, count
0
inc ax
1
mov count, ax
Поток 2
1
mov ax, count
1
inc ax
2
mov count, ax
Результат: 2

4.

Простой пример
Значение count
Поток 1
0
mov ax, count
0
inc ax
1
1
Поток 2
mov ax, count
mov count, ax
1
inc ax
1
mov count, ax
Результат: 1

5.

Простой пример
Значение count
Поток 1
Поток 2
0
mov ax, count
mov ax, count
0
inc ax
inc ax
1
mov count, ax
mov count, ax
1
Результат: 1

6.

FAQ
• Но я запустил программу 10 раз и она выдала
правильный результат, значит ошибки нет?
– Есть, но нужно запустить 11 раз.
• «inc count» на уровне ассемблера (т.е. без
регистра) мог бы решить проблему?
– Нет!
• Индивидуальный count для каждого потока?
– Да! (но не для всех задач это эффективно)

7.

Простое решение
1 int count = 0;
2 mutex mtx
3 #pragma omp parallel for
4 for (int i = 0; I < n; ++i)
if ( a[i] ==0 ){
5
lock_guard< mutex> lock(mtx);
6
7
count ++;
8
}

8.

Еще один плохой вариант
1 int count = 0;
2 mutex mtx
3 #pragma omp parallel for
4 for (int i = 0; I < n; ++i){
lock_guard< mutex> lock(mtx);
5
if ( a[i] ==0 )
6
7
count ++;
8 }

9.

Еще одно простое решение
1 atomic<int> count = 0;
2 #pragma omp parallel for
3 for (int i = 0; I < n; ++i)
if ( a[i] ==0 )
4
5
count ++;

10.

Синхронизация
• Причины необходимости синхронизации
– Параллельное выполнение операций и/или
переключение потоков
– Изменение порядка операций относительно
того, который задал программист
(оптимизации компилятора и оптимизации в
процессоре)
• Семафоры, мьютексы, события –
медленны.
• Атомарные операции не универсальны

11.

Примитивы синхронизации
• Mutex+Cond
• Semaphore
• Critical section
• Spinlock

12.

Атомарные операции
• Чтение/запись: атомарны (обычно) для
выровненных данных
• Чтение-изменение-запись:
– CompareAndSwap (x86, Itanium, Sparc)
– LoadLinked/StoreConditional (ARM, MIPS, Alpha,
PowerPC)

13.

Compare And Swap
1. bool CAS(int* data, int expected, int value)
2. {
3.
if (*data == expected) {
4.
*data = value;
5.
return true;
6.
7.
8. }
} else
return false;

14.

Compare And Swap
• Аппаратно реализованная операция
• Обычно, есть аппаратные специализации:
atomicInc, atomicAdd,…
• Но если нет – их легко сделать.
• А вот сделать LL/LC через atomic – тяжело
• Поэтому, CAS стандартизован в С++ и
поддерживается многими языками

15.

Атомарный инкремент
1. void atomicInc(int *data)
2. {
3.
int value, new_value;
4.
do{
5.
value = *data;
6.
new_value = value + 1
7.
8. }
} while ( !CAS(data, value, new_value) );

16.

Специфическая проблема CAS
• Совпадение значений не гарантирует
неизменность данных
• ABA проблема:
– Поток 1 прочитал значение A
– Поток 2 изменил значение A на B, что то сделал, а
по окончании работы вернул исходное значение A
– Поток 1 делает успешный CAS, но логика
программы нарушена
• Пример: вставка в список

17.

Load Linked / Store Conditional
• Используется аппаратный флаг изменения
данных (на уровне кеша)
• LL читает данные и устанавливает флаг
• SC меняет данные, если флаг установлен
• Любая операция записи сбрасывает флаг
• Легко сделать CAS, но не наоборот

18.

CAS из LL/SC
1. bool CAS(int* data, int expected, int value)
2. {
3.
4.
5.
6. }
if (LL(data) == expected) {
return SC(data, value);
return false;

19.

False Sharing
• флаг «изменения» относится к целой строке
кеша (как минимум, 64 байта).
• В результате, в худшем случае 100% SC
неуспешны
• Лечится выравниванием данных по
размеру линии кеша.

20.

Атомарные операции в C++ 11
1. #include <atomic>
2. std::atomic<int> sharedValue(0);
3. sharedValue.store(1);
4. sharedValue ++
compare_exchange_weak,
compare_exchange_strong …

21.

C#
System.Threading.Interlocked.Increment(ref int)
System.Threading.Interlocked.CompareExchange(
ref data, int expected, int value)
System.Threading.Interlocked.Increment(ref int)
….

22.

Python
• threading.Lock

23.

Барьеры памяти
1. void thread1()
2. {
3.
a = 1;
4.
b = 1;
5. }
6.
7. void thread2()
8. {
9.
if( b == 1 )
10.
if (a != 1)
11.
system(“format c:”);
12. }

24.

Барьеры памяти
1. void thread1()
2. {
3.
a = 1;
4.
wmb();
5.
b = 1;
6. }
7.
8. void thread2()
9. {
10.
if( b == 1 ) {
11.
rmb();
12.
if (a != 1)
13.
system(“format c:”);
14.
}
15. }

25.

Барьеры памяти
• Load/Load – все предыдущие операции
чтения завершатся перед следующим.
• Store/Store – все предыдущие операции
записи завершатся перед следующей
• Load/Store – все предыдущие чтения
завершатся перед следующей записи
• Store/Load – все предыдущие записи
завершатся перед следующим чтением

26.

Еще немного о volatile
• volatile в С++: запрет кеширования
переменной
• volatile в MS VC++ и C#: запрет кеширования
и барьер памяти

27.

Неблокирующий стек
1.
2.
3.
bool push( value_type& val )
{
back_off bkoff;
4.
5.
6.
7.
8.
9.
10.
11.
12. }
value_type * t = m_Top.load(std::memory_order_relaxed);
while ( true ) {
val.m_pNext.store( t, std::memory_order_relaxed );
if (m_Top.compare_exchange_weak(t, &val,
std::memory_order_release, std::memory_order_relaxed))
return true;
bkoff();
}

28.

Неблокирующий стек
1.
2.
3.
4.
5.
6.
7.
8.
value_type * pop()
{
back_off bkoff;
typename gc::Guard guard; // Hazard pointer guard
while ( true ) {
value_type * t = guard.protect( m_Top );
if ( t == nullptr )
return nullptr ; // stack is empty
9.
10.
11.
12.
13.
14.
15. }
value_type * pNext = t->m_pNext.load(std::memory_order_relaxed);
if ( m_Top.compare_exchange_weak( t, pNext,
std::memory_order_acquire, std::memory_order_relaxed ))
return t;
bkoff();
}

29.

FAQ 2
• Функциональные языки свободны от проблем
синхронизации?
– Теоретически, да. Чистое ФП не имеет побочных
эффектов, а значит не имеет общих переменных, а
значит и не имеет проблем синхронизации.
Только вот на чистом ФП писать не все удобно,
часто приходится мешать его с императивным
программированием.
• В моем любимом языке нет всего этого, значит
и нет проблем синхронизации?
– Если есть потоки, разделяющие общие ресурсы,
значит есть и проблемы.

30.

Библиотеки..
• С++: boost, libCds,…
• C#:
SpinLock, SpinWait, SemaphoreSlim, Concurr
entQueue<T>, ConcurrentStack<T>