UNIX: разработка сетевых приложений

  7 int clifd[MAXNCLI], iget, iput;
  8 pthread_mutex_t clifd_mutex;
  9 pthread_cond_t clifd_cond;

Определение массива для записи дескрипторов присоединенных сокетов

6-9 Мы определяем массив clifd, в который главный поток записывает дескрипторы присоединенных сокетов. Свободные потоки из пула получают по одному дескриптору из этого массива и обрабатывают соответствующий запрос, iput— это индекс в данном массиве для очередного элемента, записываемого в него главным потоком, a iget— это индекс очередного элемента массива, передаваемого свободному потоку для обработки. Разумеется, эта структура данных, совместно используемая всеми потоками, должна быть защищена, и поэтому мы используем условную переменную и взаимное исключение.

В листинге 30.25 показана функция main.

Листинг 30.25. Функция main для сервера с предварительным порождением потоков

 //server/serv08.c
  1 #include "unpthread.h"
  2 #include "pthread08.h"

  3 static int nthreads;
  4 pthread_mutex_t clifd_mutex = PTHREAD_MUTEX_INITIALIZER;
  5 pthread_cond_t clifd_cond = PTHREAD_COND_INITIALIZER;

  6 int
  7 main(int argc, char **argv)
  8 {
  9 int i, listenfd, connfd;
 10 void sig_int(int), thread_make(int);
 11 socklen_t addrlen, clilen;
 12 struct sockaddr *cliaddr;

 13 if (argc == 3)
 14 listenfd = Tcp_listen(NULL, argv[1], &addrlen);
 15 else if (argc == 4)
 16 listenfd = Tcp_listen(argv[1], argv[2], &addrlen);
 17 else
 18 err_quit("usage: serv08 [ <host> ] <port#> <#threads>");
 19 cliaddr = Malloc(addrlen);

 20 nthreads = atoi(argv[argc - 1]);
 21 tptr = Calloc(nthreads, sizeof(Thread));
 22 iget = iput = 0;

 23 /* создание всех потоков */
 24 for (i = 0; i < nthreads; i++)
 25 thread_make(i); /* завершается только основной поток */

 26 Signal(SIGINT, sig_int);

 27 for (;;) {
 28 clilen = addrlen;
 29 connfd = Accept(listenfd, cliaddr, &clilen);

 30 Pthread_mutex_lock(&clifd_mutex);
 31 clifd[iput] = connfd;
 32 if (++iput == MAXNCLI)
 33 iput = 0;
 34 if (iput == iget)
 35 err_quit("iput = iget = %d", iput);
 36 Pthread_cond_signal(&clifd_cond);
 37 Pthread_mutex_unlock(&clifd_mutex);
 38 }
 39 }

Создание пула потоков

23-25 Функция thread_make создает все потоки.

Ожидание прихода клиентского соединения

27-38 Основной поток блокируется в вызове функции accept, ожидая появления нового соединения. При появлении этого соединения дескриптор присоединенного сокета записывается в следующий элемент массива clifd после блокирования взаимного исключения. Мы также следим, чтобы индекс iget не совпал со значением индекса iput, что укажет на недостаточно большой размер массива. Условная переменная сигнализирует о прибытии нового запроса, и взаимное исключение разблокируется, позволяя одному из потоков пула обслужить прибывший запрос.

Функции thread_make и thread_main показаны в листинге 30.26. Первая из них идентична функции, приведенной в листинге 30.23.

Листинг 30.26. Функции thread_make и thread_main

 //server/pthread08.c
  1 #include "unpthread.h"
  2 #include "pthread08.h"

  3 void
  4 thread_make(int i)
  5 {
  6 void *thread_main(void*);

  7 Pthread_create(&tptr[i].thread_tid, NULL, &thread_main, (void*)i);
  8 return; /* завершается основной поток */
  9 }

 10 void*
 11 thread_main(void *arg)
 12 {
 13 int connfd;
 14 void web_child(int);

 15 printf("thread %d starting\n", (int)arg);
 16 for (;;) {
 17 Pthread_mutex_lock(&clifd_mutex);
 18 while (iget == iput)
 19 Pthread_cond_wait(&clifd_cond, &clifd_mutex);
 20 connfd = clifd[iget]; /* присоединенный сокет, который требуется
  обслужить */
 21 if (++iget == MAXNCLI)
 22 iget = 0;
 23 Pthread_mutex_unlock(&clifd_mutex);
 24 tptr[(int)arg].thread_count++;

 25 web_child(connfd); /* обработка запроса */
 26 Close(connfd);
 27 }
 28 }

Ожидание присоединенного сокета, который требует обслуживания