QNX/UNIX: Анатомия параллелизма

 #include <sys/neutrino.h>
 int SyncMutexEvent(sync_t* sync, struct sigevent* event);
 int SyncMutexEvent_r(sync_t* sync, struct sigevent* event);

Данная функция предназначена для установки обработчика ситуации, когда мьютекс попадает в состояние DEAD(то есть перераспределяется память, из которой произошел захват мьютекса). Захватить мьютекс, оказавшийся в состоянии DEAD, можно далее с помощью вызова функции SyncMutexRevive.

Ошибки выполнения функции:

EAGAIN— в данный момент ядро не имеет ресурсов для обработки запроса;

EFAULT— ошибка произошла при попытке обращения к sync;

EINVAL— объект синхронизации, на который указывает sync, не существует.

Пример применения мьютекса

Модернизируем наш пример из раздела, посвященного использованию семафора для случая множества потоков источников и приемников данных. Проблема заключается в том, что когда несколько потоков одновременно попытаются вызвать функцию push или pop, может произойти сильная путаница, поэтому код этих функций должен исполняться эксклюзивно, только одним потоком. Решить эту проблему можно двумя способами: воспользоваться бинарным семафором или мьютексом. Мы решили применить именно мьютекс и ниже расскажем причину, по которой мы здесь смешали в одной конструкции эти два элемента синхронизации.

 /* Шаблонный класс очереди данных */
 template <class T> class CDataQueue {
 public:
  CDataQueue { pthread_mutex_init(&_mutex, NULL); }
  ~CDataQueue { pthread_mutex_destroy(&_mutex); }
  void push(T _new_data) {
   pthread_mutex_lock(&_mutex);
  data_queue.push(_new_data);
  data_event.reset;
   pthread_mutex_unlock(&_mutex);
  }
  T pop {
  data_event.wait;
   pthread_mutex_lock(&_mutex);
  T res = data_queue.front;
  data_queue.pop;
   pthread_mutex_unlock(&_mutex);
  return res;
  }
 private:
  std::queue<T> data_queue;
  event data_event;
  pthread_mutex_t _mutex;
 };

На первый взгляд задача очевидна: надо не допустить одновременного исполнения двух участков кода. Почему же не воспользоваться семафором, как мы описывали, когда рассказывали о способах его применения? Дело в том, что мы хотели получить универсальное средство передачи данных между потоками, не зависящее от допущений о приоритетах потоков и степени их зависимости. Когда мы строим систему реального времени, вопрос взаимного неявного влияния разных потоков на выполнение друг друга становится очень важным. Мы уже неоднократно упоминали эффект инверсии приоритетов и те способы, которыми можно ее избежать, используя мьютекс для защиты эксклюзивно используемого кода.

Сравнение и эффективность

В этом месте временно прервем наше последовательное повествование: мы закончили рассмотрение двух наиболее известных, важных и применяемых примитивов синхронизации — семафора и мьютекса. Теперь сделаем короткую остановку и проведем их взаимное сравнение, а также попробуем на примерах оценить затраты процессорной производительности, требуемые этими механизмами.

Дело в том, что на первый взгляд эти два механизма в высшей степени подобны, особенно если речь заходит о бинарном семафоре, принимающем значения счетчика 0 либо 1. Настолько подобны, что и в обсуждениях, и даже в неспециальной литературе можно встретить утверждения, что это «одно и то же». Сейчас мы увидим, что эти два сходных механизма различаются всем: и затратами процессорного времени на их обслуживание, и целями и задачами, которые они призваны решать, и временем жизни… Начнем с простой оценки затрат процессорного времени на обслуживание каждого из механизмов, после чего остальные различия станут нам намного понятнее.

Для проведения таких оценок используем уже применявшуюся нами схему «симметричных» тестов. Почему именно их? Да, здесь нам не требуются в явном виде обменные операции потоков, но воспользуемся «симметричными» тестами просто в силу минимальных переделок того, что уже было написано ранее. Итак, первый вариант теста для мьютекса ( файл sy20m.cc):

Скоростные показатели мьютекса

 unsigned long N = 1000;
 static pthread_barrier_t bstart;
 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
 static bool debug = false;
 static char* str;
 static volatile int ind = 0;

 void* threadfunc(void* data) {
  pthread_barrier_wait(&bstart);
  unsigned long i = 0;
  char tid[8];
  sprintf(tid, "%d", pthread_self);
  uint64_t t = 0, t1;
  while (i++ != N) {
  t1 = ClockCycles;
  pthread_mutex_lock(&mutex);
  if (debug) str[ind++] = *tid;
  pthread_mutex_unlock(&mutex);
  t += ClockCycles - t1;
  sched_yield;
  }
  cout << pthread_self << "\t: cycles - "
  << t << ", on mutex - " << t / N << endl;
  return NULL;
 }

 int main(int argc, char *argv[]) {
  int opt, val;
  while ((opt = getopt(argc, argv, "n,v")) != -1) {
  switch (opt) {
  case 'n':
  if (sscanf(optarg, "%i", &val) != 1)
  cout << "parse command line error" << endl, exit(EXIT_FAILURE);
  if (val > 0) N = val;
  break;
  case 'v':
  debug = true;
  break;
  default:
  exit(EXIT_FAILURE);
  }
  }