Синхронизация потоков семафорами

#include <stdlib.h>

#include <stdio.h>

#include <string,h>

#linclude <inttypes.h>

#include <iostream.h>

#include <unistd.h>

#include <pthread.h>

#include <errno.h>

#include <semaphore.h>

unsigned long n = 1000;

static sem_t* sem; static bool debug = false;

static char* str; // строка диагностики

static volatile int ind = 0; uint64_t *t;

void* threadfunc (void* data) { unsigned long i = 0; char tid[ 8 ];

sprintf(tid, W, pthread_self());

// временная метка начала во всех потоках устанавливается //на время //достижения этой точки в последнем (активном) потоке if((int)data == Т - 1)

uint64_t с - ClockCyclesO;

for(int i = 0; i < T; i++) t[ i ] = c

// рабочий цикл переключений за счет синхронизации while(i++ < N)

sem_wait(sem + (int)data);

if(debug) str[ ind++ ] = *tid;

sem_post(sem + ((int)data +1) % T)

};

t[ (int)data ] = ClockCyclesO - t[ (int)data ]; return NULL;

};

int main(int argc, char *argv[]) { int opt, val;

while ((opt = getopt(argc, argv, "n:t: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 t':

if(sscanf(optarg, &val)!= 1)

cout «"parse command line error" «endl, exit(EXIT_FAILURE);

if(val > 0) T = val;

break; case 'v':

debug = true;

break; default:

exit(EXIT_FAILURE);

}

};

if(debug) str = new char[ T * n + 1 ]; pthread_t* tid = new pthread_t[ T ];

sem = new sem_t[ T J;

t = new uint64_t[ T ];

for(int i = 0: i < T; i++)

// все потоки, кроме последнего, будут заблокированы

// на своих семафорах сразу же после старта

if(sem_init(sem + i, 0, (i == (T - 1))? 1: 0))

perror("semaphore init"), exit(EXIT_FAILURE); if(pthread_create(tid + i, NULL,

threadfunc, (void*)i)!= EOK) perror("thread create error"), exit(EXIT_FAILURE);

};

for(int i = 0; i < T; i++)

pthread_join(tid[ i ], NULL); for(int i = 0; i < T; i++) sem_destroy(sem + i); delete [] sem;

for(int i = 0; i < T; i++)

cout «tid[ i ] << "\t: cycles - " << t[ i ]<< ";\ton semaphore - "

«t[ i ] / T / N

«endl;

delete [3 tid;

delete [] t;

if(debug) {

str[ ind ] = '\0';

cout «str «endl; delete [] str;

};

exit(EXIT_SUCCESS);

};

Логически приложение изменилось следующим образом:

Теперь у нас может быть не 2 идентичных (симметричных) потока, а произвольное их количество (ключ -t при запуске приложения).

Потоки синхронизируются не на одном семафоре - введен массив семафоров по числу потоков: каждый поток блокируется на «своем» семафоре, но разблокирует его (после очередного выполнения своего фрагмента) семафор заблокированного «соседа».

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

Из кода исключены какие бы то ни было средства принудительной передачи управления (sched_yield()) - все управление логикой ветвления осуществляется только состояниями семафоров.

Посмотрим, что у нас получилось. Запускаем приложение с диагностическим выводом идентификаторов потоков (ключ -v; он у нас был в тестах и ранее, только мы о нем не упоминали):

# nice -n-19 sy2l -n20 -t12 -v

: cycles - 664874; on semaphore - 2770

: cycles - 649150; on semaphore - 2704
: cycles - 638906; on semaphore – 2662

; cycles - 622987; on semaphore - 2595

: cycles - 611781; on semaphore - 2549

; cycles - 594515; on semaphore - 2477

: cycles - 571003; on semaphore - 2379
; cycles - 552834; on semaphore - 2303

: cycles - 536817; on semaphore - 2236
: cycles - 519357; on semaphore - 2163

; cycles - 500388; on semaphore – 2084

; cycles - 296633; on semaphore - 1235

D23456789ABCD23456789ABCD23456789ABCD23456789ABCD23456789ABCD23456789ABCD234 56789ABCD23456789ABC023456789ABCD23456789ABCD23456789ABCD23456789ABCD2345678 9ABCD23456789ABCD23456789ABCD23456789ABCD23456789ABCD23456789ABC023456789ABC D23456789A8C

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

# nice -n-19 sy2l -	пЮОООО -t12
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -

# nice 2

tt nice.-n-19 sy21 -пЮОООО -t7

	cycles	- 859768389	on	semaphore -
	cycles	- 859756956	on	semaphore -
	cycles	- 859745649	on	semaphore -
	cycles	- 859736698	on	semaphore -
	cycles	- 859724685	on	semaphore -
	cycles	- 859707720	on	semaphore -
	cycles	- 859554045	on	semaphore -

on semaphore - 1281-n-19 sy21 -n50000 -t13: cycles - 832789852;

з	cycles -			on	semaphore -
	cycles -			on	semaphore -
	cycles -			on	semaphore -
	cycles -			on	semaphore -
	cycles -			on	semaphore -
8 '	cycles -			on	semaphore -
	cycles -			on	semaphore -
	cycles -			On	semaphore -
	cycles -			on	semaphore -
	cycles -			on	semaphore -
	cycles -			on	semaphore -
	cycles -			on	semaphore -
# nice -	i-19 sy21	-П50000 -t17
	cycles -		on	semaphore -
	cycles -			on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -	114302151E		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cydes -		on	semaphore -
	cycles -			on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -
	cycles -		on	semaphore -

Есть некоторая корреляция времени переключения контекста с размером выборки и количеством обрабатывающих потоков, но она в широком диапазоне этих параметров не превышает 8%. В данном приложении эта численная величина включает в себя: блокирование на семафоре, переключение на контекст другого потока и разблокирование семафора. Если вспомнить, что раньше мы получали оценки для принудительного (посредством sched_yield()) переключения контекста потоков в 375 процессорных циклов, а для захвата-освобождения семафора - порядка 870, то эти цифры хорошо согласуются с полученными сейчас результатами.

Рассматриваемые примитивы служат принципиально различным целям. Мьютекс, как уже было сказано ранее, предназначен в первую очередь для регламентации доступа к участкам программного кода. Семафоры же больше предназначены для регламентации порядка доступа к определенным объектам данных. Классическими задачами этого класса являются задачи «производитель - потребитель», когда М производителей создают некоторые объекты данных (читая эти данные с реальных внешних устройств, или создавая их как результат только внутренних вычислений, или любым другим способом), а N потребителей независимо берут произведенные объекты данных на последующую обработку.

Это настолько общий и часто встречающийся класс задач, что покажем для него простейший «скелет» в виде отдельного приложения, в котором отслеживание порядка доступа потребителей будет осуществлять счетный семафор (файл sy22.cc). Для простоты понимания приложение сделано как трансформация кода предшествующей группы тестов. В качестве имитации производства объекта данных, как и в качестве его обработки потребителем, используется пассивная пауза (delayO) на случайную величину (производство и обработка объектов данных в коде не показаны, так как это не относится к существу рассматриваемого - нас интересуют процессы синхронизации этих операций, а не сами операции).

Кроме основной нашей цели это приложение дополнительно демонстрирует:

Практическое использование принудительного завершения (отмены) потоков «извне» с управлением состоянием завершаемости потоков и расстановкой точек отмены, о чем мы уже говорили ранее.

Использование атомарных (непрерываемых) операций (например, atomic_add_value()), о которых мы будем говорить чуть позже.

Использование реентерабельных форм функций стандартной библиотеки, безопасных в многопоточной среде (rand_г() вместо rand()).

Дата добавления: 2014-12-10