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

Пример

Пример

Свяжем теперь всю эту информацию воедино в примере. Мы начнем с функции main нашего клиента UDP, представленного в листинге 8.3, и изменим в ней только номер порта с SERV_PORT на 7 (стандартный эхо-сервер, см. табл. 2.1).

В листинге 22.4 показана функция dg_cli. Единственное изменение по сравнению с листингом 8.4 состоит в замене вызовов функций sendto и recvfrom вызовом нашей новой функции dg_send_recv.

Перед тем как представить функцию dg_send_recv и наши функции RTT, которые она вызывает, мы показываем в листинге 22.5 нашу схему реализации функциональных свойств, повышающих надежность клиента UDP. Все функции, имена которых начинаются с rtt_, описаны далее.

Листинг 22.4. Функция dg_cli, вызывающая нашу функцию dg_send_recv

//rtt/dg_cli.c
 1 #include "unp.h"
 2 ssize_t Dg_send_recv(int, const void*, size_t, void*, size_t,
 3  const SA*, socklen_t);
 4 void
 5 dg_cli(FILE *fp, int sockfd, const SA *pservaddr, socklen_t servlen)
 6 {
 7  ssize_t n;
 8  char sendline[MAXLINE], recvline[MAXLINE + 1];
 9  while (Fgets(sendline, MAXLINE, fp) != NULL) {
10   n = Dg_send_recv(sockfd, sendline, strlen(sendline),
11    recvline, MAXLINE, pservaddr, servlen);
12   recvline[n] = 0; /* завершающий нуль */
13   Fputs(recvline, stdout);
14  }
15 }

Листинг 22.5. Схема функций RTT и последовательность их вызова

static sigjmp_buf jmpbuf;
{
 формирование запроса
 signal(SIGALRM, sig_alrm); /* устанавливаем обработчик сигнала */
 rtt_newpack(); /* инициализируем значение счетчика rexmt нулем */
sendagain:
 sendto();
 alarm(rtt_start()); /* задаем аргумент функции alarm равным RTO */
 if (sigsetjmp(jmpbuf, 1) != 0) {
  if (rtt_timeout()) /* удваиваем RTO, обновляем оценочные значения */
   отказываемся от дальнейших попыток
  goto sendagain; /* повторная передача */
 }
 do {
  recvfrom();
 } while (неправильный порядковый номер);
 alarm(0); /* отключаем сигнал alarm */
 rtt_stop(); /* вычисляем RTT и обновляем оценочные значения */
 обрабатываем ответ
}
void sig_alrm(int signo) {
 siglongjmp(jmpbuf, 1);
}

Если приходит ответ, но его порядковый номер отличается от предполагаемого, мы снова вызываем функцию recvfrom, но не отправляем снова тот же запрос и не перезапускаем работающий таймер повторной передачи. Обратите внимание, что в крайнем правом случае на рис. 22.2 последний ответ, полученный на отправленный повторно запрос, будет находиться в приемном буфере сокета до тех пор, пока клиент не решит отправить следующий запрос (и получить на него ответ). Это нормально, поскольку клиент прочитает этот ответ, отметит, что порядковый номер отличается от предполагаемого, проигнорирует ответ и снова вызовет функцию recvfrom.

Мы вызываем функции sigsetjmp и siglongjmp, чтобы предотвратить возникновение ситуации гонок с сигналом SIGALRM, который мы описали в разделе 20.5. В листинге 22.6 показана первая часть нашей функции dg_send_recv.

Листинг 22.6. Функция dg_send_recv: первая половина

//rtt/dg_send_recv.c
 1 #include "unprtt.h"
 2 #include <setjmp.h>
 3 #define RTT_DEBUG
 4 static struct rtt_info rttinfo;
 5 static int rttinit = 0;
 6 static struct msghdr msgsend, msgrecv;
   /* предполагается, что обе структуры инициализированы нулем */
 7 static struct hdr {
 8  uint32_t seq; /* порядковый номер */
 9  uint32_t ts;  /* отметка времени при отправке */
10 } sendhdr, recvhdr;
11 static void signalrm(int signo);
12 static sigjmp_buf jmpbuf;
13 ssize_t
14 dg_send_recv(int fd, const void *outbuff, size_t outbytes,
15  void *inbuff, size_t inbytes,
16  const SA *destaddr, socklen_t destlen)
17 {
18  ssize_t n;
19  struct iovec iovsend[2], iovrecv[2];
20  if (rttinit == 0) {
21   rtt_init(&rttinfo); /* первый вызов */
22   rttinit = 1;
23   rtt_d_flag = 1;
24  }
25  sendhdr.seq++;
26  msgsend.msg_name = destaddr;
27  msgsend.msg_namelen = destlen;
28  msgsend.msg_iov = iovsend;
29  msgsend.msg_iovlen = 2;
30  iovsend[0].iov_base = &sendhdr;
31  iovsend[0].iov_len = sizeof(struct hdr);
32  iovsend[1].iov_base = outbuff;
33  iovsend[1].iov_len = outbytes;
34  msgrecv.msg_name = NULL;
35  msgrecv.msg_namelen = 0;
36  msgrecv.msg_iov = iovrecv;
37  msgrecv.msg_iovlen = 2;
38  iovrecv[0].iov_base = &recvhdr;
39  iovrecv[0].iov_len = sizeof(struct hdr);
40  iovrecv[l].iov_base = inbuff;
41  iovrecv[l].iov_len = inbytes;
1-5 Мы включаем новый заголовочный файл unprtt.h, показанный в листинге 22.8, который определяет структуру rtt_info, содержащую информацию RTT для клиента. Мы определяем одну из этих структур и ряд других переменных.

6-10 Мы хотим скрыть от вызывающего процесса добавление порядкового номера и отметки времени в начало каждого пакета. Проще всего использовать для этого функцию writev, записав свой заголовок (структура hdr), за которым следуют данные вызывающего процесса, в виде одной дейтаграммы UDP. Вспомните, что результатом выполнения функции writev на дейтаграммном сокете является отправка одной дейтаграммы. Это проще, чем заставлять вызывающий процесс выделять для нас место в начале буфера, а также быстрее, чем копировать наш заголовок и данные вызывающего процесса в один буфер (под который мы должны выделить память) для каждой функции sendto. Но поскольку мы работаем с UDP и нам необходимо задать адрес получателя, следует использовать возможности, предоставляемые структурой iovec функций sendmsg и recvmsg и отсутствующие в функциях sendto и recvfrom. Вспомните из раздела 14.5, что в некоторых системах доступна более новая структура msghdr, включающая вспомогательные данные (msg_control), тогда как в более старых системах вместо них применяются элементы msg_accright (так называемые права доступа — access rights), расположенные в конце структуры. Чтобы избежать усложнения кода директивами #ifdef для обработки этих различий, мы объявляем две структуры msghdr как static. При этом они инициализируются только нулевыми битами, а затем неиспользованные элементы в конце структур просто игнорируются.

20-24 При первом вызове нашей функции мы вызываем функцию rtt_init.

25-41 Мы заполняем две структуры msghdr, используемые для ввода и вывода. Для данного пакета мы увеличиваем на единицу порядковый номер отправки, но не устанавливаем отметку времени отправки, пока пакет не будет отправлен (поскольку он может отправляться повторно, а для каждой повторной передачи требуется текущая отметка времени).

Вторая часть функции вместе с обработчиком сигнала sig_alarm показана в листинге 22.7.

Листинг 22.7. Функция dg_send_recv: вторая половина

//rtt/dg_send_rеcv.c
42  Signal(SIGALRM, sig_alrm);
43  rtt_newpack(&rttinfo); /* инициализируем для этого пакета */
44 sendagain:
45  sendhdr.ts = rtt_ts(&rttinfo);
46  Sendmsg(fd, &msgsend, 0);
47  alarm(rtt_start(&rttinfo)); /* вычисляем тайм-аут. запускаем таймер */
48  if (sigsetjmp(jmpbuf, 1) != 0) {
49   if (rtt_timeout(&rttinfо) < 0) {
50    err_msg("dg_send_recv: no response from server, giving up");
51    rttinit = 0; /* повторная инициализация для следующего вызова */
52    errno = ETIMEDOUT;
53    return (-1);
54   }
55   goto sendagain;
56  }
57  do {
58   n = Recvmsg(fd, &msgrecv, 0);
59  } while (n < sizeof(struct hdr) || recvhdr.seq != sendhdr.seq);
60  alarm(0); /* останавливаем таймер SIGALRM */
61  /* вычисляем и записываем новое значение оценки RTT */
62  rtt_stop(&rttinfo, rtt_ts(&rttinfo) — recvhdr.ts);
63  return (n - sizeof(struct hdr)); /* возвращаем размер полученной
                                        дейтаграммы */
64 }
65 static void
66 sig_alrm(int signo)
67 {
68  siglongjmp(jmpbuf, 1);
69 }

42-43 Для сигнала SIGALRM устанавливается обработчик сигналов, а функция rtt_newpack устанавливает счетчик повторных передач в нуль.

45-47 Функция rtt_ts получает текущую отметку времени. Отметка времени хранится в структуре hdr, которая добавляется к данным пользователя. Одиночная дейтаграмма UDP отправляется функцией sendmsg. Функция rtt_start возвращает количество секунд для этого тайм-аута, а сигнал SIGALRM контролируется функцией alarm.

48 Мы устанавливаем буфер перехода для нашего обработчика сигналов с помощью функции sigsetjmp. Мы ждем прихода следующей дейтаграммы, вызывая функцию recvmsg. (Совместное использование функций sigsetjmp и siglongjmp вместе с сигналом SIGALRM мы обсуждали применительно к листингу 20.5.) Если время таймера истекает, функция sigsetjmp возвращает 1.

49-55 Когда возникает тайм-аут, функция rtt_timeout вычисляет следующее значение RTO (используя экспоненциальное смещение) и возвращает -1, если нужно прекратить попытки передачи дейтаграммы, или 0, если нужно выполнить очередную повторную передачу. Когда мы прекращаем попытки, мы присваиваем переменной errno значение ETIMEDOUT и возвращаемся в вызывающую функцию.

57-59 Мы ждем прихода дейтаграммы, вызывая функцию recvmsg. Длина полученной дейтаграммы не должна быть меньше размера структуры hdr, а ее порядковый номер должен совпадать с порядковым номером запроса, ответом на который предположительно является эта дейтаграмма. Если при сравнении хотя бы одно из этих условий не выполняется, функция recvmsg вызывается снова.

60-62 Когда приходит ожидаемый ответ, функция alarm отключается, а функция rtt_stop обновляет оценочное значение RTT. Функция rtt_ts возвращает текущую отметку времени, и отметка времени из полученной дейтаграммы вычитается из текущей отметки, что дает в результате RTT.

65-69 Вызывается функция siglongjmp, результатом выполнения которой является то, что функция sigsetjmp в dg_send_recv возвращает 1.

Теперь мы рассмотрим различные функции RTT, которые вызывались нашей функцией dg_send_recv. В листинге 22.8 показан заголовочный файл unprtt.h.

Листинг 22.8. Заголовочный файл unprtt.h

//lib/unprtt.h
 1 #ifndef __unp_rtt_h
 2 #define __unp_rtt_h
 3 #include "unp.h"
 4 struct rtt_info {
 5  float    rtt_rtt;    /* последнее измеренное значение RTT в секундах */
 6  float    rtt_srtt;   /* сглаженная оценка RTT в секундах */
 7  float    rtt_rttvar; /* сглаженные средние значения отклонений
                            в секундах */
 8  float    rtt_rto;    /* текущее используемое значение RTO, в секундах */
 9  int      rtt_nrexmt; /* количество повторных передач: 0, 1, 2, ... */
10  uint32_t rtt_base;   /* число секунд, прошедшее после 1.1.1970 в начале */
11 };
12 #define RTT_RXTMIN    2 /* минимальное значение тайм-аута для
                              повторной передачи, в секундах */
13 #define RTT_RXTMAX   60 /* максимальное значение тайм-аута для
                              повторной передачи, в секундах */
14 #define RTT_MAXNREXMT 3 /* максимально допустимое количество
                              повторных передач одной дейтаграммы */
15 /* прототипы функций */
16 void     rtt_debug(struct rtt_info*);
17 void     rtt_init(struct rtt_info*);
18 void     rtt_newpack(struct rtt_info*);
19 int      rtt_start(struct rtt_info*);
20 void     rtt_stop(struct rtt_info*, uint32_t);
21 int      rtt_timeout(struct rtt_info*);
22 uint32_t rtt_ts(struct rtt_info*);
23 extern int rtt_d_flag; /* может быть ненулевым при наличии
                             дополнительной информации */
24 #endif /* _unp_rtt_h */

4-11 Эта структура содержит переменные, необходимые для того, чтобы определить время передачи пакетов между клиентом и сервером. Первые четыре переменных взяты из уравнений, приведенных в начале этого раздела.

12-14 Эти константы определяют минимальный и максимальный тайм-ауты повторной передачи и максимальное число возможных повторных передач.

В листинге 22.9 показан макрос RTT_RTOCALC и первые две из четырех функций RTT.

Листинг 22.9. Макрос RTT_RTOCALC, функции rtt_minmax и rtt_init

//lib/rtt.c
 1 #include "unprtt.h"
 2 int rtt_d_flag = 0; /* отладочный флаг; может быть установлен в
                          ненулевое значение вызывающим процессом */
 3 /* Вычисление значения RTO на основе текущих значений:
 4  * сглаженное оценочное значение RTT + четырежды сглаженная
 5  * величина отклонения.
 6  */
 7 #define RTI_RTOCALC(ptr) ((ptr)->rtt_srtt + (4.0 * (ptr)->rtt_rttvar))
 8 static float
 9 rtt_minmax(float rto)
10 {
11  if (rto < RTT_RXTMIN)
12   rto = RTT_RXTMIN;
13  else if (rto > RTT_RXTMAX)
14   rto = RTT_RXTMAX;
15  return (rto);
16 }
17 void
18 rtt_init(struct rtt_info *ptr)
19 {
20  struct timeval tv;
21  Gettimeofday(&tv, NULL);
22  ptr->rtt_base = tv.tv_sec; /* количество секунд, прошедших с 1.1.1970 */
23  ptr->rtt_rtt = 0;
24  ptr->rtt_srtt = 0;
25  ptr->rtt_rttvar = 0.75;
26  ptr->rtt_rto = rtt_minmax(RTT_RTOCALC(ptr));
27  /* первое RTO (srtt + (4 * rttvar)) = 3 с */
28 }
3-7 Макрос вычисляет RTO как сумму оценочной величины RTT и оценочной величины среднего отклонения, умноженной на четыре.

8-16 Функция rtt_minmax проверяет, что RTO находится между верхним и нижним пределами, заданными в заголовочном файле unprtt.h.

17-28 Функция rtt_init вызывается функцией dg_send_recv при первой отправке пакета. Функция gettimeofday возвращает текущее время и дату в той же структуре timeval, которую мы видели в функции select (см. раздел 6.3). Мы сохраняем только текущее количество секунд с момента начала эпохи Unix, то есть с 00:00:00 1 января 1970 года (UTC). Измеряемое значение RTT обнуляется, а сглаженная оценка RTT и среднее отклонение принимают соответственно значение 0 и 0,75, в результате чего начальное RTO равно 3 с (4?0,75).

В листинге 22.10 показаны следующие три функции RTT.

Листинг 22.10. Функции rtt_ts, rtt_newpack и rtt_start

//lib/rtt.c
34 uint32_t
35 rtt_ts(struct rtt_info *ptr)
36 {
37  uint32_t ts;
38  struct timeval tv;
39  Gettimeofday(&tv, NULL);
40  ts = ((tv.tv_sec - ptr->rtt_base) * 1000) + (tv.tv_usec / 1000);
41  return (ts);
42 }
43 void
44 rtt_newpack(struct rtt_info *ptr)
45 {
46  ptr->rtt_nrexmt = 0;
47 }
48 int
49 rtt_start(struct rtt_info *ptr)
50 {
51  return ((int)(ptr->rtt_rto + 0.5)); /* округляем float до int */
52  /* возвращенное значение может быть использовано как аргумент
       alarm(rtt_start(&fоо)) */
53 }
34-42 Функция rtt_ts возвращает текущую отметку времени для вызывающего процесса, которая должна содержаться в отправляемой дейтаграмме в виде 32-разрядного целого числа без знака. Мы получаем текущее время и дату из функции gettimeofday и затем вычитаем число секунд в момент вызова функции rtt_init (значение, хранящееся в элементе rtt_base структуры rtt_info). Мы преобразуем это значение в миллисекунды, а также преобразуем в миллисекунды значение, возвращаемое функцией gettimeofday в микросекундах. Тогда отметка времени является суммой этих двух значений в миллисекундах.

Разница во времени между двумя вызовами функции rtt_ts представляется количеством миллисекунд между этими двумя вызовами. Но мы храним отметки времени в 32-разрядном целом числе без знака, а не в структуре timeval.

43-47 Функция rtt_newpack просто обнуляет счетчик повторных передач. Эта функция должна вызываться всегда, когда новый пакет отправляется в первый раз.

48-53 Функция rtt_start возвращает текущее значение RTO в миллисекундах. Возвращаемое значение затем может использоваться в качестве аргумента функции alarm.

Функция rtt_stop, показанная в листинге 22.11, вызывается после получения ответа для обновления оценочного значения RTT и вычисления нового значения RTO.

Листинг 22.11. Функция rtt_stop: обновление показателей RTT и вычисление нового

//lib/rtt.c
62 void
63 rtt_stop(struct rtt_info *ptr, uint32_t ms)
64 {
65  double delta;
66  ptr->rtt_rtt = ms / 1000.0; /* измеренное значение RTT в секундах */
67  /*
68   * Обновляем оценочные значения RTT среднего отклонения RTT.
69   * (См. статью Джекобсона (Jacobson). SIGCOMM'88. Приложение А.)
70   * Здесь мы для простоты используем числа с плавающей точкой.
71   */
72  delta = ptr->rtt_rtt - ptr->rtt_srtt;
73  ptr->rtt_srtt += delta / 8; /* g - 1/8 */
74  if (delta < 0.0)
75   delta = -delta; /* |delta| */
76  ptr->rtt_rttvar += (delta - ptr->rtt_rttvar) / 4; /* h - 1/4 */
77  ptr->rtt_rto = rtt_minmax(RTT_RTOCALC(ptr));
78 }
62-78 Вторым аргументом является измеренное RTT, полученное вызывающим процессом при вычитании полученной в ответе отметки времени из текущей (функция rtt_ts). Затем применяются уравнения, приведенные в начале этого раздела, и записываются новые значения переменных rtt_srtt, rtt_rttvar и rtt_rto.

Последняя функция, rtt_timeout показана в листинге 22.12. Эта функция вызывается, когда истекает время таймера повторных передач.

Листинг 22.12. Функция rtt_timeout: применение экспоненциального смещения

//lib/rtt.c
83 int
84 rtt_timeout(struct rtt_info *ptr)
85 {
86  ptr->rtt_rto *= 2; /* следующее значение RTO */
87  if (++ptr->rtt_nrexmt > RTT_MAXNREXMT)
88   return (-1); /* закончилось время, отпущенное на попытки отправить
                     этот пакет */
89  return (0);
90 }
86 Текущее значение RTO удваивается — в этом и заключается экспоненциальное смещение.

87-89 Если мы достигли максимально возможного количества повторных передач, возвращается значение -1, указывающее вызывающему процессу, что дальнейшие попытки передачи должны прекратиться. В противном случае возвращается 0.

В нашем примере клиент соединялся дважды с двумя различными эхо-серверами в Интернете утром рабочего дня. Каждому серверу было отправлено по 500 строк. По пути к первому серверу было потеряно 8 пакетов, по пути ко второму — 16. Один из потерянных шестнадцати пакетов, предназначенных второму серверу, был потерян дважды, то есть пакет пришлось дважды передавать повторно, прежде чем был получен ответ. Все остальные потерянные пакеты пришлось передать повторно только один раз. Мы могли убедиться, что эти пакеты были действительно потеряны, посмотрев на выведенные порядковые номера каждого из полученных пакетов. Если пакет лишь опоздал, но не был потерян, после повторной передачи клиент получает два ответа: соответствующий запоздавшему первому пакету и повторно переданному. Обратите внимание, что у нас нет возможности определить, что именно было потеряно (и привело к необходимости повторной передачи клиентского запроса) — сам клиентский запрос или же ответ сервера, высланный после получения такого запроса.

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