Atomic Lock

[Igors]

Добрый день

Хочу сделать такую штуку на QAtomic

Код

C++ (Qt)
QAtomicInt state;
 
void AcquireRead( void )
{
 while (true) {
  if (state >= 1024)  {    // старший бит взведен - данные недоступны
   DoNothing();             // холостой ход 
   continue;              // крутимся в while (рано или поздно др. нитка сбросит старший бит)
  }
  state.fetchAndAddAcquire(1); // атомарно увеличиваем счетчик (еще 1 нитка имеет доступ к данным)
  break;                        // доступ получен
 }
}
 

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

Код

C++ (Qt)
  if (state >= 1024)  
                            <---- здесь пороется собака
  state.fetchAndAddAcquire(1); 
 

А как сделать правильно?

Спасибо

[lit-uriy]

QMutexLocker?

[SABROG]

QMutexLocker?

Он хочет написать lock-free алгоритм, который будет грузить ядро процессора на 100%, но не блокировать другие потоки. Мутексы с семафорами для этих целей не подходят.

Приблизительно так можно сделать.

Код

C++ (Qt)
while(true) {
    int cachedstate = state.fetchAndStoreAcquire((int)state); // кэшируем значение глобальной переменной
    if (cachedstate < 1024 
        && state.testAndSetRelaxed(cachedstate, cachedstate)) { // проверяем не изменилось ли значение пока переходили на следующую инструкцию другим потоком и одновременно проверяем меньше ли оно 1024
        state.fetchAndAddRelease(1); // увеличиваем счетчик на 1 и выходим из цикла
        break;
    }
/*
Если значение больше или не совпадает со значением в глобальной переменной, значит проиграли race-condition другому потоку и позволяем ему завершить начатое (то есть какой-то из потоков сейчас внутри блока if)
*/
    QThread::yieldCurrentThread(); // DoNothing() / continue
}
 

Или так (побыстрее будет на первоначальную проверку):

Код

C++ (Qt)
while(true) {
    int cachedstate = state.fetchAndStoreAcquire((int)state);
    if (cachedstate >= 1024)
        QThread::yieldCurrentThread(); // DoNothing() / continue
    else if (state.testAndSetRelaxed(cachedstate, cachedstate)) {
        state.fetchAndAddRelease(1);
        break;
    }
}
 

[Igors]

Он хочет написать lock-free алгоритм, который будет грузить ядро процессора на 100%, но не блокировать другие потоки. Мутексы с семафорами для этих целей не подходят.

Совершенно верно   Однако предложенные Вами варианты страдают тем же дефектом что и начальный

Код

C++ (Qt)
    if (cachedstate < 1024 
        && state.testAndSetRelaxed(cachedstate, cachedstate)) 
 // if вернул true но до того как следующая строка выполнится  др. нитка может установить state  = 1024                
        state.fetchAndAddRelease(1);
 

Код

C++ (Qt)
if (state.testAndSetRelaxed(cachedstate, cachedstate)) 
                    <--  здесь пороется собака
        state.fetchAndAddRelease(1);
 

[SABROG]

Можно так сделать:

Код

C++ (Qt)
while(true) {
    int cachedstate = state.fetchAndStoreAcquire((int)state);
    if (cachedstate >= 1024)
        QThread::yieldCurrentThread(); // DoNothing() / continue
    else if (state.testAndSetRelaxed(cachedstate, cachedstate+1))
        break;
}
 

[Igors]

Код

C++ (Qt)
while(true) {
    int cachedstate = state.fetchAndStoreAcquire((int)state);
    if (cachedstate >= 1024)
        QThread::yieldCurrentThread(); // DoNothing() / continue
    else if (state.testAndSetRelaxed(cachedstate, cachedstate+1))
        break;
}
 

Думаю что идея правильная, но надо отрихтовать

1) state.fetchAndStoreAcquire((int)state); Если значение было изменено между (int) state и fetchAndStoreAcquire, то эти изменения будут убиты. Надо заменить на просто = (QAtomicInt имеет оператор int)

2) testAndSetRelaxed обещает что значение будет установлено, но не гарантирует что др. нитка не вклинится в нашу оборону. Поэтому надо заменить на testAndSetAcquire

3) yieldCurrentThread - дело темное устроит ли это по скорости, вообще неясно как "прокручиваться" наилучшим образом. Так что выделим это  в DoNothing

С учетом вышеизложенного

Код

C++ (Qt)
void AcquireRead( QAtomicInt & state )
{
 while(true) {
     int cachedState = state;
     if (cachedState >= 1024)
         DoNothing();
     else 
       if (state.testAndSetAcquire(cachedState, cachedState + 1))
         break;
 }
}
 

Другие мнения?

[SABROG]

1) state.fetchAndStoreAcquire((int)state); Если значение было изменено между (int) state и fetchAndStoreAcquire, то эти изменения будут убиты. Надо заменить на просто = (QAtomicInt имеет оператор int)

Изменения могут быть, но это лишь приведет к одному "холостому" циклу. QAtomicInt не имеет атомарных конструктора и оператора int(), аналогов std::atomic<>::load()/store() в нём нет.

2) testAndSetRelaxed обещает что значение будет установлено, но не гарантирует что др. нитка не вклинится в нашу оборону. Поэтому надо заменить на testAndSetAcquire

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

3) yieldCurrentThread - дело темное устроит ли это по скорости, вообще неясно как "прокручиваться" наилучшим образом. Так что выделим это  в DoNothing

Алгоритм где я применял этот метод был в 2 раза быстрей boost'овского. К тому же это позволяет не расходывать процессорное время на холостые ходы, отдаем ресурсы потоку, который в этом нуждается.

С учетом вышеизложенного
     int cachedState = state;
     if (cachedState >= 1024)

Тут без fetchAndStoreAcquire() можно запросто получить оптимизацию компилятора или re-ordering.

[Igors]

Изменения могут быть, но это лишь приведет к одному "холостому" циклу.

Не могу с Вами согласиться

Код:

state.fetchAndStoreAcquire((int)state)
// псевдо-ассеблер :-)

push [state];    // текущее значение state записывается в стек (более вероятно - в регистр)
call [fetchAndStoreAcquire];   // перезапись 

Пусть на стек было помещено значение 0. Но перед вызовом fetchAndStoreAcquire др. нитка успела установить это значение в 1. Результат: инкремент/декремент (баланс счетчика) нарушен

Тут без fetchAndStoreAcquire() можно запросто получить оптимизацию компилятора или re-ordering.

Да, но здесь это приведет лишь к еще 1 холостому циклу. "Проскочившее" значение будет отсечено дальше

[SABROG]

Не могу с Вами согласиться

Код

C++ (Qt)
// предположим state 1023, пока мы его передавали в fetchAndStoreAcquire оно стало 1024
int cachedstate = state.fetchAndStoreAcquire((int)state); // cachedstate стало 1024
if (cachedstate >= 1024) // сработало условие (хотя не должно было), просто передали управление другому потоку, который продолжит цикл с прерванного места (возможно второй круг цикла или несколько лишних инструкций)
 
// другой вариант. state 1024, пока мы его передавали в fetchAndStoreAcquire оно как-то поменялось на 1
int cachedstate = state.fetchAndStoreAcquire((int)state); // cachedstate стало 1
if (cachedstate >= 1024) // не срабатывает, уходим на else
...
else if (state.testAndSetRelaxed(cachedstate, cachedstate+1)) // и state и cachedstate равны 1, производим нужные действия. 
 

Да, но здесь это приведет лишь к еще 1 холостому циклу. "Проскочившее" значение будет отсечено дальше

Компилятор может это оптимизировать как-нибудь так убрав "лишнюю" переменную:

Код

C++ (Qt)
 while(true) {
     if (state >= 1024)
         DoNothing();
     else 
       if (state.testAndSetAcquire(state, state + 1))
         break;
 }
 

[Igors]

Код

C++ (Qt)
// другой вариант. state 1024, пока мы его передавали в fetchAndStoreAcquire оно как-то поменялось на 1
int cachedstate = state.fetchAndStoreAcquire((int)state); // cachedstate стало 1
if (cachedstate >= 1024) // не срабатывает, уходим на else
...
else if (state.testAndSetRelaxed(cachedstate, cachedstate+1)) // и state и cachedstate равны 1, производим нужные действия. 
 

Нет, state и cachedstate  не равны, ведь fetchAndStoreAcquire сработал точно и теперь state = 1024 (если его не долбанули по вертикали)

Код

C++ (Qt)
while(true) {
     if (state >= 1024)
         DoNothing();
     else 
       if (state.testAndSetAcquire(state, state + 1))
         break;
 }
 

Значение QAtomicInt (_q_value) объявлено volatile, так что не должен. А хоть бы и поместил state на регистр - чем нам это плохо? Та же локальная переменная что и хотели

[SABROG]

Нет, state и cachedstate  не равны, ведь fetchAndStoreAcquire сработал точно и теперь state = 1024 (если его не долбанули по вертикали)

Что значит "долбанули по вертикали"?

Значение QAtomicInt (_q_value) объявлено volatile, так что не должен. А хоть бы и поместил state на регистр - чем нам это плохо? Та же локальная переменная что и хотели

Вопрос в том скопирует ли оптимизатор компилятора переменную в стек/регистер или полезет прямиком в память по адресу. Ведь volatile переменные изначально использовались для получения данных с устройства, например:

Код

C++ (Qt)
volatile int& port = некий адрес в памяти, на который мэппиться устройство;
int a = port; // получили первые 4 байта
int b = port; // получили следующие 4 байта
... // и т.д. компилятор каждый раз лезет в память и ничего нигде не кеширует и не сохраняет при этом, так как при повторном чтении уже пользователем данные потеряются
 

[Igors]

Что значит "долбанули по вертикали"?

т.е. асинхронно, из др нитки или по прерыванию

Вопрос в том скопирует ли оптимизатор компилятора переменную в стек/регистер или полезет прямиком в память по адресу. Ведь volatile переменные изначально использовались для получения данных с устройства, например:

Я так понимаю что volatile переменная должна всегда честно читаться/писаться "прямиком".

[Igors]

Возможен и др вариант решения

Код

C++ (Qt)
void AcquireRead( QAtomicInt & state )
{
 while(true) {
     if (state >= 1024)
         DoNothing();
     else {
         int result = state.fetchAndAddAcquire(1);  // атомарно увеличиваем счетчик
         if (result >= 1024)                       // если старший бит успел проскочить до инкремента
           state.fetchAndAddAcquire(-1);           // откат
         else
           break;                                  // доступ  получен
    }
 }
}

Имеется ввиду что счетчик в младших битах - скромное число ниток которое никогда не достигнет 1024 (бит блокировки)