Камерон Хьюз - Параллельное и распределенное программирование на С++

// Листинг 4.5. Подход 1: скелет программы реализации II модели управляющего и рабочих потоков

//...

pthread_mutex_t Mutex = PTHREAD_MUTEX_INITIALIZER

int AvailableThreads

pthread_t Thread[Max_Threads]

void decrementThreadAvailability(void)

void incrementThreadAvailability(void)

int threadAvailability(void);

// boss thread

{

//...

if(sysconf(_SC_THREAD_THREADS_MAX) > 0){

AvailableThreads = sysconf(_SC_THREAD_THREADS_MAX)

}

else{

AvailableThreads = Default

}

int Count = 1;

loop while(Request Queue is not empty)

if(threadAvailability()){

Count++

decrementThreadAvailability()

classify request

switch(request type)

{

case X : pthread_create(&(Thread[Count])...taskX...)

case Y : pthread_create(&(Thread[Count])...taskY...)

case Z : pthread_create(&(Thread[Count])...taskZ...)

//...

}

}

else{

//free up thread resources

}

end loop

}

void *taskX(void *X)

{

// process X type request

incrementThreadAvailability()

return(NULL)

}

void *taskY(void *Y)

{

// process Y type request

incrementThreadAvailability()

return(NULL)

}

void *taskZ(void *Z)

{

// process Z type request

decrementThreadAvailability()

return(NULL)

}

В листинге 4.5 управляющий поток динамически создает поток для обработки каждого нового запроса, который поступает в систему. Однако существует ограничение на количество потоков (максимальное число потоков), которое можно создать в процессе. Для обработки n типов запросов существует n задач. Чтобы гарантировать, что максимальное число потоков на процесс не будет превышено, определяются следующие дополнительные функции:

threadAvailability()

incrementThreadAvailability()

decrementThreadAvailability()

В листинге 4.6 содержится псевдокод реализации этих функций.

// Листинг 4.6. Функции, которые управляют возможностью

// создания потоков

void incrementThreadAvailability(void)

{

//...

pthread_mutex_lock(&Mutex)

AvailableThreads++

pthread_mutex_unlock(&Mutex)

}

void decrementThreadAvailability(void)

{

//...

pthread_mutex_lock(&Mutex)

AvailableThreads—

pthread_mutex_unlock(&Mutex)

}

int threadAvailability(void)

{

//...

pthread_mutex_lock(&Mutex)

if(AvailableThreads > 1)

return 1

else

return 0

pthread_mutex_unlock(&Mutex)

}

Ф ункция threadAvailability()возвратит число 1, если максимально допустимое количество потоков для процесса еще не достигнуто. Эта функция опрашивает глобальную переменную ThreadAvailability,в которой хранится число потоков, еще доступных для процесса. Управляющий поток вызывает функцию decrementThreadAvailability(),которая декрементирует эту глобальную переменную до создания им рабочего потока. Каждый рабочий поток вызывает функцию incrementThreadAvailability(),которая инкрементирует глобальную переменную ThreadAvailabilityдо начала его выполнения. Обе функции содержат обращение к функции pthread_mutex_lock() до получения доступа к этой глобальной переменной и обращение к функции pthread_mutex_unlock()после него. Если максимально допустимое количество потоков превышено, управляющий поток может отменить создание потока, если это возможно, или породить другой процесс, если это необходимо. Функции taskX(), taskY() и taskZ() выполняют код, предназначенный для обработки запроса соответствующего типа.

Другой подход к реализации модели делегирования состоит в создании управляющим потоком пула потоков, которым (вместо создания под каждый новый запрос нового потока) переназначаются новые запросы. Управляющий поток во время инициализации создает некоторое количество потоков, а затем каждый созданный поток приостанавливается до тех пор, пока в очередь не будет добавлен новый запрос. Управляющий поток для выделения запросов из очереди по-прежнему использует цикл событий. Но вместо создания нового потока для обслуживания очередного запроса, управляющий поток уведомляет уже существующий поток о необходимости обработки запроса. Этот подход к реализации модели делегирования представлен в листинге 4.7.

// Листинг 4.7. Подход 2: скелет программы реализации . модели управляющего и рабочих потоков

pthread_t Thread[N]

// boss thread

{

pthread_create(&(Thread[1]...taskX...);

pthread_create(&(Thread[2]...taskY...);

pthread_create(&(Thread[3]...taskZ...);

//...

loop while(Request Queue is not empty

get request

classify request

switch(request type)

{

case X :

enqueue request to XQueue

signal Thread[1]

case Y :

enqueue request to YQueue

signal Thread[2]

case Z :