Я пытаюсь реализовать высокопроизводительную очередь блокировки, поддерживаемую циклическим буфером поверх pthreads, semaphore.h и gcc atomic buildins.Очередь должна обрабатывать несколько одновременных программ чтения и записи из разных потоков.
Я выделил какое-то состояние гонки, и я не уверен, является ли это ошибочным предположением о поведении некоторых атомарных операций.и семафоры, или мой дизайн в корне ошибочен.
Я извлек и упростил его до приведенного ниже автономного примера.Я ожидаю, что эта программа никогда не вернется.Однако он возвращает после нескольких сотен тысяч итераций с повреждением, обнаруженным в очереди.
В приведенном ниже примере (для демонстрации) он на самом деле ничего не хранит, он просто устанавливает 1 ячейку, которая будет содержатьфактические данные и 0 для представления пустой ячейки.Существует счетный семафор (вакансий), представляющий количество вакантных ячеек, и другой счетный семафор (жителей), представляющий количество занятых ячеек.
Авторы делают следующее:
- декрементвакансии
- атомно получить следующий индекс заголовка (размер очереди мода)
- записать в него
- увеличить число жителей
Читатели делают наоборот:
- декременты-обитатели
- атомарно получить следующий хвостовой индекс (размер очереди мода)
- прочитать из него
- увеличить вакансии
Я ожидаю, что с учетом вышеизложенного, точно один поток может одновременно читать или записывать любую данную ячейку.
Любые идеи о том, почему она не работает или стратегии отладки приветствуются.Код и вывод ниже ...
#include <stdlib.h>
#include <semaphore.h>
#include <iostream>
using namespace std;
#define QUEUE_CAPACITY 8 // must be power of 2
#define NUM_THREADS 2
struct CountingSemaphore
{
sem_t m;
CountingSemaphore(unsigned int initial) { sem_init(&m, 0, initial); }
void post() { sem_post(&m); }
void wait() { sem_wait(&m); }
~CountingSemaphore() { sem_destroy(&m); }
};
struct BlockingQueue
{
unsigned int head; // (head % capacity) is next head position
unsigned int tail; // (tail % capacity) is next tail position
CountingSemaphore vacancies; // how many cells are vacant
CountingSemaphore occupants; // how many cells are occupied
int cell[QUEUE_CAPACITY];
// (cell[x] == 1) means occupied
// (cell[x] == 0) means vacant
BlockingQueue() :
head(0),
tail(0),
vacancies(QUEUE_CAPACITY),
occupants(0)
{
for (size_t i = 0; i < QUEUE_CAPACITY; i++)
cell[i] = 0;
}
// put an item in the queue
void put()
{
vacancies.wait();
// atomic post increment
set(__sync_fetch_and_add(&head, 1) % QUEUE_CAPACITY);
occupants.post();
}
// take an item from the queue
void take()
{
occupants.wait();
// atomic post increment
get(__sync_fetch_and_add(&tail, 1) % QUEUE_CAPACITY);
vacancies.post();
}
// set cell i
void set(unsigned int i)
{
// atomic compare and assign
if (!__sync_bool_compare_and_swap(&cell[i], 0, 1))
{
corrupt("set", i);
exit(-1);
}
}
// get cell i
void get(unsigned int i)
{
// atomic compare and assign
if (!__sync_bool_compare_and_swap(&cell[i], 1, 0))
{
corrupt("get", i);
exit(-1);
}
}
// corruption detected
void corrupt(const char* action, unsigned int i)
{
static CountingSemaphore sem(1);
sem.wait();
cerr << "corruption detected" << endl;
cerr << "action = " << action << endl;
cerr << "i = " << i << endl;
cerr << "head = " << head << endl;
cerr << "tail = " << tail << endl;
for (unsigned int j = 0; j < QUEUE_CAPACITY; j++)
cerr << "cell[" << j << "] = " << cell[j] << endl;
}
};
BlockingQueue q;
// keep posting to the queue forever
void* Source(void*)
{
while (true)
q.put();
return 0;
}
// keep taking from the queue forever
void* Sink(void*)
{
while (true)
q.take();
return 0;
}
int main()
{
pthread_t id;
// start some pthreads to run Source function
for (int i = 0; i < NUM_THREADS; i++)
if (pthread_create(&id, NULL, &Source, 0))
abort();
// start some pthreads to run Sink function
for (int i = 0; i < NUM_THREADS; i++)
if (pthread_create(&id, NULL, &Sink, 0))
abort();
while (true);
}
Скомпилируйте вышеприведенный код следующим образом:
$ g++ -pthread AboveCode.cpp
$ ./a.out
Вывод каждый раз отличается, но вот один пример:
corruption detected
action = get
i = 6
head = 122685
tail = 122685
cell[0] = 0
cell[1] = 0
cell[2] = 1
cell[3] = 0
cell[4] = 1
cell[5] = 0
cell[6] = 1
cell[7] = 1
Моя система - Ubuntu 11.10 на Intel Core 2:
$ uname -a
Linux 3.0.0-14-generic #23-Ubuntu SMP \
Mon Nov 21 20:28:43 UTC 2011 x86_64 x86_64 x86_64 GNU/Linux
$ cat /proc/cpuinfo | grep Intel
model name : Intel(R) Core(TM)2 Quad CPU Q9300 @ 2.50GHz
$ g++ --version
g++ (Ubuntu/Linaro 4.6.1-9ubuntu3) 4.6.1
Спасибо, Эндрю.