Книга: Программирование для Linux. Профессиональный подход

Листинг 4.12. (job-queue3.c) Работа с очередью заданий с применением семафора

#include <malloc.h>
#include <pthread.h>
#include <semaphore.h>
struct job {
 /* Ссылка на следующий элемент связанного списка. */
 struct job* next;
 /* Другие поля, описывающие требуемую операцию... */
};
/* Список отложенных заданий. */
struct job* job_queue;
/* Исключающий семафор, защищающий очередь. */
pthread_mutex_t job_queue_mutex =
 PTHREAD_MUTEX_INITIALIZER;
/* Семафор, подсчитывающий число гаданий в очереди. */
sem_t job_queue_count;
/* Начальная инициализация очереди. */
void initialize_job_queue() {
 /* Вначале очередь пуста. */
 job_queue = NULL;
 /* Устанавливаем начальное значение счетчика семафора
    равным 0. */
 sem_init(&job_queue_count, 0, 0);
}
/* Обработка заданий до тех пор, пока очередь не опустеет. */
void* thread_function(void* arg) {
 while (1) {
  struct job* next_job;
  /* Дожидаемся готовности семафора. Если его значение больше
     нуля, значит, очередь не пуста; уменьшаем счетчик на 1.
     В противном случае операция блокируется до тех пор, пока
     в очереди не появится новое задание. */
  sem_wait(&job_queue_count);
  /* Захват исключающего семафора, защищающего очередь. */
  pthread_mutex_lock(&job_queue_mutex);
  /* Мы уже знаем, что очередь не пуста, поэтому без лишней
     проверки запрашиваем новое задание. */
  next_job = job_queue;
  /* Удаляем задание из списка. */
  job_queue = job_queue->next;
  /* освобождаем исключающий семафор, так как работа с
     очередью окончена. */
  pthread_mutex_unlock(&job_queue_mutex);
  /* Выполняем задание. */
  process_job(next_job);
  /* Очистка. */
  free(next_job);
 }
 return NULL;
}
/* Добавление нового задания в начало очереди. */
void enqueue_job(/* Передача необходимых данных... */) {
 struct job* new_job;
 /* Выделение памяти для нового объекта задания. */
 new_job = (struct job*)malloc(sizeof(struct job));
 /* Заполнение остальных полей структуры JOB... */
 /* Захватываем исключающий семафор, прежде чем обратиться
    к очереди. */
 pthread_mutex_lock(&job_queue_mutex);
 /* Помещаем новое задание в начало очереди. */
 new_job->next = job_queue;
 job_queue = new_job;
 /* Устанавливаем семафор, сообщая о том, что в очереди появилось
    новое задание. Если есть потоки, заблокированные в ожидании
    семафора, один из них будет разблокирован и
    обработает задание. */
 sem_post(&job_queue_count);
 /* Освобождаем исключающий семафор. */
 pthread_mutex_unlock(&job_queue_mutex);
}

Прежде чем извлекать задание из очереди, каждый поток дожидается семафора. Если счетчик семафора равен нулю, т.е. очередь пуста, поток блокируется до тех пор, пока в очереди не появится новое задание и счетчик не станет положительным.

Функция enqueue_job() добавляет новое задание в очередь. Подобно потоковой функции, она захватывает исключающий семафор, перед тем как обратиться к очереди. После добавления задания функция enqueue_job() устанавливает семафор, сообщая потокам о том, что задание доступно. В программе, показанной в листинге 4.12, потоки никогда не завершаются: если задания не поступают в течение длительного времени, все потоки переводятся в режим блокирования функцией sem_wait().

Оглавление книги