Использование блокировок чтения/записи

Использование блокировок чтения/записи

Построим приложение, использующее блокировку чтения/записи (файл 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.

# nice -n-19 sy11 500 .2

evaluation time: 1.24973 sec.

# nice -n-19 sy12 500 .2

evaluation time: 0.440904 sec.

При «жесткой» блокировке мы не получаем никакого выигрыша за счет параллельного выполнения запросов к данным, а при использовании блокировки чтения/записи — 3-кратный выигрыш. Проделаем то же самое, но в условиях гораздо меньшей интенсивности обновления данных относительно общего потока запросов:

# nice -n-19 sy10 500 .02

evaluation time 0.989699 sec.

# nice -n-19 sy11 500 .02

evaluation time 0.98391 sec.

# nice -n-19 sy12 500 .02

evaluation time 0.0863443 sec.

Выигрыш становится более чем 10-кратным.

Показанные примеры (sy10.cc, sy11.cc, sy12.cc) в высшей степени условны: картина происходящего будет существенно другой при замене пассивного ожидания (delay()) на активные процессорные операции над данными, но общие тенденции сохраняются.