AtomicData или механизм атомарной операции записи данных

[Akon]

В этом случае читатель может получить неконсистентные данные.

Прошу показать каким образом.

Пусть в структуре (Data) будет два члена и она меняется из состояния {0, 0} в состояние {1, 1}. Сценарий:
1. Писатель выполнил a = other.a; из directAssignment для m_data1; m_data1 = {1, 0}.
2. Читатель выполнил const Data result = m_data1; result = {1, 0}.
3. Писатель завершил первое присваивание (m_data1) и во втором присваивании (из directAssignment, а не из reverseAssignment - ваш вариант) выполнил a = other.a для m_data2; m_data2 = {1, 0}.
4. Читатель проверяет условие while (result != m_data2), и мы имеем result == m_data2 == {1, 0} - неконсистентность! Data должна меняться атомарно для читателей, т.е. либо {0, 0}, либо {1, 1}.

и они начали одновременно первый писать в m_data1, второй читать из него. В чём тогда синхронизация?

Синхронизация в том, что читатель получит корректные данные (независимо от временной внутренней неконсистентности).

Структура Data просто определяет атомарную группу, в самой структуре данные-члены понимаются атомарными в смысле присваивания. Прошу прощения, что не указал этого.

[Igors]

Edit: кстати, shallow копирование довольно сложно реализовать полностью thread safe. Вы уверены, что QString, например, не падает при одновременном чтении и присвоении?

Да, они подсовывают указатель на новые данные атомарно. Так что при "чтении и присвоении" все нормально, также и при "записи и присвоении" - но не при "записи и чтении"

3. Писатель завершил первое присваивание (m_data1) и во втором присваивании (из directAssignment, а не из reverseAssignment - ваш вариант) выполнил a = other.a для m_data2; m_data2 = {1, 0}.
4. Читатель проверяет условие while (result != m_data2), и мы имеем result == m_data2 == {1, 0} - неконсистентность! Data должна меняться атомарно для читателей, т.е. либо {0, 0}, либо {1, 1}.

Согласен. Тогда может лучше так

Код

C++ (Qt)
	const Data data() const
	{
		const Data result = m_data1; /*B*/
 
		while (result != m_data1 || result != m_data2)
			result = m_data1;
 
		return result;
	}
 

Чтобы обойтись без утомительного сравнения "задом наперед"

[brankovic]

Структура Data просто определяет атомарную группу, в самой структуре данные-члены понимаются атомарными в смысле присваивания. Прошу прощения, что не указал этого.

хорошо, тогда как с этим:

data == AA AA
W хочет записать BB BB
W записал B, data == BA AA
R хочет прочесть data
R прочёл BA, заснул
W закончил, и стал писать AA AA, записал AA A, data == AA BA
R проснулся, прочитал BA, в стеке у него BA BA, он их сравнил и остался доволен, но data никогда не принимала значения BA

[m_ax]

И так, вроде ничего не упустил)

Код

C++ (Qt)
#ifndef ATOMICDATA_H
#define ATOMICDATA_H
 
#include <assert.h>
#include <cstdatomic>
 
template <class T>
class AtomicData
{
public:
    typedef atomic<int> AtomicInt;
    AtomicData()  : m_count_readers(0), m_state_write(-1) {}
 
    void setData(const T &data) {
        while (!acquireWrite(m_count_readers, m_state_write));
        m_data = data;
        releaseWrite(m_state_write);
    }
    T getData() const {
        while (!acquireRead(m_count_readers, m_state_write));
        T data = m_data;
        releaseRead(m_count_readers, m_state_write);
        return data;
    }
private:
    AtomicInt m_count_readers;
    AtomicInt m_state_write;
    T m_data;
 
    static bool acquireWrite(AtomicInt &readers, AtomicInt &state_write) {
 
        bool tmp = (++state_write == 0);
        state_write = 0;
 
        return (tmp && !readers);
    }
    //-------------------------------------------------------------------------
 
    static void releaseWrite(AtomicInt &state_write) {
        state_write = -1;
    }
    //-------------------------------------------------------------------------
 
    static bool acquireRead(AtomicInt &readers, AtomicInt &state_write) {
 
        if (!readers)
            readers = (state_write == -1);
        else
            readers++;
 
        return readers;
    }
    //-------------------------------------------------------------------------
 
    static void releaseRead(AtomicInt &readers, AtomicInt &state_write) {
        if (--readers == 0)
            state_write = -1;
    }
    //-------------------------------------------------------------------------
};
 
 
#endif // ATOMICDATA_H
 

Пояснения:
m_counter_readers - число читающих потоков (в данный момент времени)
m_state_write - если = -1 - никто не пишет, в противном случае есть ожидающие записи

Читать данные могут одновременно несколько потоков, записывать всегда только какой-то один.
Если кто-то пишет, читать запрещено.

[Igors]

R проснулся, прочитал BA, в стеке у него BA BA, он их сравнил и остался доволен, но data никогда не принимала значения BA

Верно, совпадение возможно и при обратном порядке присваивания

Код

C++ (Qt)
        if (!readers)
            readers = (state_write == -1);
 

Те же грабли. Последняя строка не атомарна.

С Ларисой я действительно пил вино, но это было на пасху

Т.е. state_write действительно был -1, но до тех пор пока это присвоится readers, др. нитка успеет захватить по записи.

Лучше держать все в одном AtomicInt. с двумя переменными заморочек намного больше. Ваше упорство избежать CAS "заслуживает лучшего применения"  И как там у Вас с холостым ходом?

[m_ax]

Код

C++ (Qt)
        if (!readers)
            readers = (state_write == -1);
 

Те же грабли. Последняя строка не атомарна.

С Ларисой я действительно пил вино, но это было на пасху

Т.е. state_write действительно был -1, но до тех пор пока это присвоится readers, др. нитка успеет захватить по записи.

Лучше держать все в одном AtomicInt. с двумя переменными заморочек намного больше. Ваше упорство избежать CAS "заслуживает лучшего применения"  И как там у Вас с холостым ходом?
[/quote]
Не атомарно? Ну это можно исправить следующим образом:

Код

C++ (Qt)
static bool acquireRead(AtomicInt &readers, AtomicInt &state_write) {
 
        if (!readers)
            readers = 1; // атомарно
            int tmp = (state_write == -1); // атомарно
            readers = tmp; // атомарно
        else
            readers++;
 
        return readers;
    }
 

И как там у Вас с холостым ходом?

В смысле? С ним что-то не так?

[Igors]

Не атомарно? Ну это можно исправить следующим образом:

Код

C++ (Qt)
        if (!readers) 
            readers = 1; // атомарно
 

Опять не атомарно - ведь N ниток могли увеличить readers "между 2-мя строками" и, присваивая единицу, Вы это значение теряете

И как там у Вас с холостым ходом?

В смысле? С ним что-то не так?

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

[brankovic]

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

это как-то уж слишком коварно.. По смыслу m_ax спинлок хотел, а в спинлоке так и должно быть.

[Igors]

это как-то уж слишком коварно.. По смыслу m_ax спинлок хотел, а в спинлоке так и должно быть.

Не должно. Про свой OSX могу сказать: OSSpinLockLock (спиннер) не просто "крутит while". А TBB сначала просто делает какое-то число nop, затем вставляет pause а потом уж shed_yield. При этом учитывается специфика процессора. В общем, это на порядок сложнее чем то что сейчас обсуждается    

А без этого (увы) могут быть заметные тормоза для др. задач

[m_ax]

А что Вы скажите на это  :

Код

C++ (Qt)
static bool acquireRead(AtomicInt &readers, AtomicInt &state_write) {
 
        if (!readers) {
            int tmp = (++state_write == 0);
            readers = tmp;
        }
        else
            readers++;
 
        return readers;
    }
 

[m_ax]

это как-то уж слишком коварно.. По смыслу m_ax спинлок хотел, а в спинлоке так и должно быть.

Не должно. Про свой OSX могу сказать: OSSpinLockLock (спиннер) не просто "крутит while". А TBB сначала просто делает какое-то число nop, затем вставляет pause а потом уж shed_yield. При этом учитывается специфика процессора. В общем, это на порядок сложнее чем то что сейчас обсуждается    

А без этого (увы) могут быть заметные тормоза для др. задач

Если удасца реализовать первоночальный SpinLock, то его можно будет ускорить создав буфер даных. Тогда простои при записи и чтении можно будет существенно сократить. 

[m_ax]

Короче, окончательный вариант SpinLock:

Код

C++ (Qt)
#ifndef SPINLOCK_H
#define SPINLOCK_H
 
#include <assert.h>
#include <cstdatomic>
 
template <class T>
class SpinLock
{
public:
    typedef atomic<int> AtomicInt;
    SpinLock()  : m_count_readers(0), m_state_write(-1) {}
 
    void setData(const T &data) {
        while (!acquireWrite(m_count_readers, m_state_write));
        m_data = data;
        releaseWrite(m_state_write);
    }
    T getData() const {
        while (!acquireRead(m_count_readers, m_state_write));
        T data = m_data;
        releaseRead(m_count_readers, m_state_write);
        return data;
    }
private:
    AtomicInt m_count_readers;
    AtomicInt m_state_write;
    T m_data;
 
    static bool acquireWrite(AtomicInt &readers, AtomicInt &state_write) {
 
        bool tmp = (++state_write == 0);
        state_write = 0;
 
        return (tmp && !readers);
    }
    //-------------------------------------------------------------------------
 
    static void releaseWrite(AtomicInt &state_write) {
        state_write = -1;
    }
    //-------------------------------------------------------------------------
 
    static bool acquireRead(AtomicInt &readers, AtomicInt &state_write) {
 
        if (!readers) {
            int tmp = (++state_write == 0);
            readers = tmp;
        }
        else
            readers++;
 
        return readers;
    }
    //-------------------------------------------------------------------------
 
    static void releaseRead(AtomicInt &readers, AtomicInt &state_write) {
        if (--readers == 0)
            state_write = -1;
    }
    //-------------------------------------------------------------------------
};
 
#endif // SPINLOCK_H
 

[brankovic]

это как-то уж слишком коварно.. По смыслу m_ax спинлок хотел, а в спинлоке так и должно быть.

Не должно. Про свой OSX могу сказать: OSSpinLockLock (спиннер) не просто "крутит while". А TBB сначала просто делает какое-то число nop, затем вставляет pause а потом уж shed_yield. При этом учитывается специфика процессора. В общем, это на порядок сложнее чем то что сейчас обсуждается    

А без этого (увы) могут быть заметные тормоза для др. задач

а, понятно, думал о wait-free речь пойдёт. Никогда такую оптимизацию в действии не видел. Но интересно, что здесь сложного? Казалось бы раз в n циклов вызывать yeild, и всё?

[brankovic]

Короче, окончательный вариант SpinLock:

Код

C++ (Qt)
static bool acquireRead(AtomicInt &readers, AtomicInt &state_write)
{
   if (!readers)
   {
      int tmp = (++state_write == 0);
      readers = tmp;
   }
   else
      readers++;
 
   return readers;
}
 

R0 читал
R1 дошёл до if, увидел, что кто-то читает, отправился на else
R1 заснул
R0 ушёл (readers == 0)
W пришёл и захватил на запись
R1 проснулся, сделал ++readers и вернул true

теперь R0 читает, а W пишет

[Akon]

хорошо, тогда как с этим:

data == AA AA
W хочет записать BB BB
W записал B, data == BA AA
R хочет прочесть data
R прочёл BA, заснул
W закончил, и стал писать AA AA, записал AA A, data == AA BA
R проснулся, прочитал BA, в стеке у него BA BA, он их сравнил и остался доволен, но data никогда не принимала значения BA

Верно! Я рассмотрел синхронизацию в контексте только одного вызова setData, позабыв про возможность 2-х и более вызовов. Спасибо.

Навскидку, если добавить отслеживание вызовов (проблему переполнения счетчика пока опустил):

Код:

struct Data 
{
	int a;
	int b;
	...

	Data& directAssignment(const Data& other)
	{
		a = other.a;
		b = other.b;
		...
		return *this;
	}

	Data& reverseAssignment(const Data& other)
	{
		b = other.b;
		a = other.a;
		...
		return *this;
	}
};

class DataManager
{
public:
	// called by reader threads (several threads)
	const Data data() const
	{
		const Data result = m_data1;
		int counter = counter_;

		while (result != m_data2 || counter != counter_) {
			result = m_data1;
			counter = counter_;
		}

		return result;
	}

	// called by writer thread (single thread)
    void setData(const Data& value)
    {
		if (value == m_data1) return;

		++counter_;
		m_data1.directAssignment(value);
		m_data2.reverseAssignment(value);
    }

private:
	volatile Data m_data1;
	volatile Data m_data2;
	volatile int counter_;
}