Правильное использование атомики
Я написал небольшую облегченную очередь push / pop, основанную на векторе (считал, что она должна быть быстрой), вот так:
template <typename T> class MyVectorQueue{
public:
MyVectorQueue (size_t sz):my_queue(sz),start(0),end(0){}
void push(const T& val){
size_t idx=atomic_fetch_add(&end,1UL);
if (idx==my_queue.size())
throw std::runtime_error("Limit reached, initialize to larger vector");
my_queue[idx]=val;
}
const T& pop(){
if (start==end)
return end__;
return my_queue[start.fetch_add(1UL)];
}
size_t empty()
{
return end==start;
}
const T& end() {
return end__;
}
private:
T end__;
std::atomic<size_t> start,end;
std::vector<T> my_queue;
};
Размер вектора должен быть известен и
Мне интересно, почему это не потокобезопасно? При каких обстоятельствах это портит мою структуру?
Решение
Ваш start а также end являются атомарными переменными, но с использованием std::vector::operator[] не атомарная операция, что делает его не поточно-ориентированным.
Предположим, у вас есть 10 потоков и размер vector 5. Теперь предположим, что все они выполняются, скажем, push,
Теперь предположим, что все 10 потоков прошли проверку и if (end==my_queue.size()) оценивается в falseкак и end не достигла предела, то есть — vector не полный
Тогда возможно, что все они увеличивают end и в то же время позвонить std::vector::operator[], По крайней мере 5 потоков будут пытаться получить доступ к элементам «вне» вектора.
Другие решения
Вы используете operator[] выдвигать элементы, но это не увеличит вектор для добавления элемента. Поэтому вы получите неопределенное поведение (и, возможно, нарушение прав доступа) при попытке добавить элемент в индекс, который не существует.
Кроме того, хотя вы используете атомарную операцию на start а также end vector не атомарный. Так, например, вы могли бы вызвать несколько потоков pushони атомно меняются end а потом все звонят operator[] который не является потокобезопасным. Вместо этого вы должны подумать об использовании мьютекса и станд :: Deque :
std::mutex mutex;
std::deque<T> my_queue;
void push(const T& val){
std::lock_guard<std::mutex> guard(mutex);
//..code to check if full
my_queue.push_back(val);
}
const T& pop(){
std::lock_guard<std::mutex> guard(mutex);
//code to check if empty and that start index does not pass end index
T item=my_queue.front();
my_queue.pop_front();
return item;
}
Хотя этот код выглядит опасно неправильно на первый взгляд, на многих аккаунтах она содержит только одну проблему. В противном случае это прекрасно, в пределах своих ограничений.
Вы создаете vector инициализируется до некоторого определенного размера, и вы не разрешаете толкать больше элементов, чем данный размер. Это несколько «странное поведение», но если это то, что нужно, то проблем нет.
призвание vector::operator[] выглядит очень проблематичным для безопасности потоков, потому что это не атомарная операция, но это действительно не проблема как таковая. Все это vector::operator[] do возвращает ссылку на элемент, соответствующий указанному вами индексу. Он не выполняет никакой проверки границ или перераспределения или каких-либо других сложных вещей, которые может быть прерывание при наличии параллелизма (по всей вероятности, это сводится к чему-то вроде одной инструкции LEA).
Ты используешь fetch_add в любом случае это правильный подход, гарантирующий уникальность индексов среди потоков. Если индексы уникальны, никакие два потока никогда не получат доступ к одному и тому же элементу, поскольку не имеет значения, является ли этот доступ атомарным. Даже если несколько потоков атомарно увеличивают счетчик одновременно, все они будут получать разные результаты (то есть, никакие приращения не будут «потеряны» в пути). По крайней мере, это теория, и это верно как далеко push тоже обеспокоен
Тот самый реальный проблема заключается в pop функция где (кроме как в push) ты не атомарно fetch_add индекс (или, вернее, и то и другое индексы) в локальные переменные до их сравнения и вызова operator[],
Это проблема, так как if(start==end) не атомарен как таковой, и он не атомарен в сочетании с призванием operator[] несколько строк спустя. Вы должны получить и то и другое значения для сравнения в одной атомарной операции, или вы не можете утверждать, что сравнение имеет какое-либо значение. Иначе:
- Другой поток (или два или три потока) может увеличивать
startили жеend(или оба), пока вы сравниваете их, и ваша проверка на достижение максимальной емкости не удастся. - Другой поток может увеличиваться
startпосле этой проверки, но доfetch_addвнутри звонкаoperator[]видно, что заставляет вас индексировать вне границ и вызывает неопределенное поведение.
Неинтуитивная вещь, на которую следует обратить внимание: хотя вы делаете «правильные вещи» с помощью атомарных операций, программа все равно будет вести себя неправильно, так как не просто индивидуальный операции должны быть атомарными.