Олег Цилюрик - QNX/UNIX: Анатомия параллелизма
EOK — успешное выполнение;
EAGAIN — при первом использовании статически инициированной блокировки чтения/записи (PTHREAD_RWLOCK_INITIALIZER) недостаточно системных ресурсов для инициализации блокировки чтения/записи;
EDEADLK — вызывающий поток уже является владельцем блокировки в эксклюзивном режиме;
EFAULT — сбой при обращении ядра к rwl;
EINVAL — rwl указывает на неверный объект блокировки чтения/записи.
Функция pthread_rwlock_trywrlock() возвращает значения:
EOK — успешное выполнение;
EAGAIN — при первом использовании статически инициированной блокировки чтения/записи (PTHREAD_RWLOCK_INITIALIZER) недостаточно системных ресурсов для инициализации блокировки чтения/записи;
EBUSY — блокировка уже захвачена в режиме чтения или записи;
EDEADLK — вызывающий поток уже является владельцем блокировки в эксклюзивном режиме;
EFAULT — сбой при обращении ядра к rwl;
EINVAL — rwl указывает на неверный объект блокировки чтения/записи.
Функция pthread_rwlock_timedwrlock() возвращает значения:
EOK — успешное выполнение;
EAGAIN — система не может захватить блокировку по записи, поскольку достигнуто максимальное число блокировок по записи для данного объекта;
EDEADLK — вызывающий поток уже является владельцем блокировки в эксклюзивном режиме;
EINVAL — неверный параметр вызова: либо rwl указывает на неинициализированный объект блокировки чтения/записи, либо время тайм-аута abs задано меньше нуля или равно или выше предельного значения 1000 миллионов;
ETIMEDOUT — не удалось захватить блокировку до истечения заданного срока тайм-аута.
Освобождение блокировки
int pthread_rwlock_unlock(pthread_rwlock_t* rwl);
Функция освобождает захваченный любым образом объект блокировки чтения/записи. Если объект был захвачен в режиме множественного использования (блокировки по чтению), то количество его освобождений должно равняться количеству захватов.
Возвращаемые значения:
EOK — успешное завершение;
EAGAIN — при первом использовании статически инициированной блокировки чтения/записи (PTHREAD_RWLOCK_INITIALIZER) недостаточно системных ресурсов для инициализации блокировки чтения/записи;
EFAULT — ядро не смогло обратиться к объекту rwl;
EINVAL — объект rwl указывает на неверно инициированный объект блокировки чтения/записи;
EPERM — нет потоков, захвативших объект rwl в режиме чтения или записи, или вызывающий поток не владеет блокировкой в режиме записи.
Использование блокировок чтения/записи
Построим приложение, использующее блокировку чтения/записи (файл sy10.cc):
Эффективность блокировки чтения/записи#include <sys/syspage.h>
#include <sys/neutrino.h>
#include <list>
// сколь угодно сложные элементы внутренней базы данных
// приложения; в примере мы используем их простейший вид
typedef int element;
// база данных приложения - динамический список элементов
class dbase : public list<element> {
static const int READ_DELAY = 1, WRITE_DELAY = 2;
public:
// операция "добавить в базу данных"
void add(const element& e) {
int pos = size() * rand() / RAND_MAX;
list<element>::iterator p = begin();
for (int i = 0; i < pos; i++) p++;
insert(p, e);
delay(WRITE_DELAY);
}
// операция "найти в базе данных"
int pos(const element& e) {
int n = 0;
for (list<element>::iterator i = begin(); i != end(); i++, n++)
if (*i == e) {
delay(READ_DELAY);
break;
}
if (n == size()) n = -1;
return n;
}
} data;
inline element erand(unsigned long n) {
return (element)((n * rand()) / RAND_MAX);
}
inline bool wrand(double p) {
return (double)rand() / (double)RAND_MAX < p;
}
int main(int argc, char *argv[]) {
// общее число обращений приложения к базе данных
static unsigned long n = 1000;
// вероятность обновления базы данных при обращении
static double p = .1;
unsigned long m;
if (argc > 1 && (m = atoll(argv[1])) ! = 0) n = m;
double q;
if (argc > 2 && (q = atof(argv[2])) != 0) p = q;
// начальное заполнение базы данных
for (int i = 0; i < n; i++) data.add(erand(n));
// тактовая частота процессора (для измерения времени)
const uint64_t cps = SYSPAGE_ENTRY(qtime)->cycles_per_sec;
// последовательность n обращений к базе данных,
// из них p*n требуют обновления списка, а остальные
// (1-p)*n требуют только выборки данных:
uint64_t t = ClockCycles();
for (int i = 0; i < n; i++) {
element e = erand(n);
if (!wrand(p)) data.pos(e);
else data.add(e);
}
t = ((ClockCycles() - t) * 1000000000) / cps;
cout << "evaluation time: " << (double)t / 1.E9
<< " sec." << endl;
return EXIT_SUCCESS;
}
Перед нами простейшая последовательная программа, которая массированно читает свою базу данных и изредка ее модифицирует. Для выполнения реальных операций чтения/записи данных программе необходимо выполнять некоторые достаточно продолжительные операции. В приведенном коде эти операции имитируются задержками delay(WRITE_DELAY) и delay(READ_DELAY).
Возникает совершенно естественное желание преобразовать последовательные запросы к данным в параллельные (файл sy11.cc). Для этого преобразуем структуру списка элементов, с которым работаем:
class dbase : public list<element> {
static const int READ_DELAY = 1, WRITE_DELAY = 2;
pthread_mutex_t loc;
public:
dbase(void) { pthread_mutex_init(&loc, NULL); }
~dbase(void) { pthread_mutex_destroy(&loc); }
void add(const elements e) {
pthread_mutex_lock(&loc);
int pos = size() * rand() / RAND_MAX;
list<element>::iterator p = begin();
for (int i = 0; i < pos; i++) p++;
insert(p, e);
delay(WRITE_DELAY);
pthread_mutex_unlock(&loc);
}
int pos(const elements e) {
int n = 0;
pthread_mutex_lock(&loc);
for (list<element>::iterator i = begin(); i != end(); i++, n++)
if (*i == e) {
delay(READ_DELAY);
break;
}
pthread_mutex_unlock(&loc);
if (n == size()) n = -1;
return n;
}
} data;
А в вызывающей программе цикл запросов к данным преобразуем в:
pthread_t *h = new pthread_t[n];
uint64_t t = ClockCycles();
for (int i = 0; i < n; i++) {
element e = erand(n);
pthread_create(h + i, NULL, wrand(p) ? add : pos, (void*)e);
}
for (int i = 0; i < n; i++)
pthread_join(h[i], NULL);
t = ((ClockCycles() - t) * 1000000000) / cps;
delete h;
А используемые этим фрагментом функции потоков определим как:
static void* add(void* par) { data.add((element)par); }
static void* pos(void* par) { data.pos((element)par); }
Совершенно естественно, что список элементов, из которого мы извлекаем данные (и куда изредка помещаем новые), должен защищаться как при модификации, так и при считывании (во избежание их одновременной модификации «со стороны»). Понятно, что в представленном решении мы чересчур перестраховались: во время считывания мы должны защищаться от потенциальной одновременной модификации, но нет необходимости защищать структуру данных от параллельного считывания. Поэтому переопределим структуру данных (файл sy12.cc), используя блокировку чтения/записи, оставив все прочее без изменений:
class dbase : public list<element> {
static const int READ_DELAY = 1, WRITE_DELAY = 2;
pthread_rwlock_t loc;
public:
dbase(void) { pthread_rwlock_init(&loc, NULL); }
~dbase(void) { pthread_rwlock_destroy(&loc); }
void add(const elements e) {
pthread_rwlock_wrlock(&loc);
int pos = size() * rand() / RAND_MAX;
list<element>::iterator p = begin();
for (int i = 0; i < pos; i++) p++;
insert(p, e);
delay(WRITE_DELAY);
pthread_rwlock_unlock(&loc);
}
int pos(const elements e) {
int n = 0;
pthread_rwlock_rdlock(&loc);
for (list<element>::iterator i = begin(); i != end(); i++, n++)
if (*i == e) {
delay(READ_DELAY);
break;
}
pthread_rwlock_unlock(&loc);
if (n == size()) n = -1;
return n;
}
} data;
А теперь пришло время сравнить варианты:
# nice -n-19 sy10 500 .2
evaluation time: 1.2296 sec.
