UDP 非阻塞 client/server 中生成的随机 EWOULDBLOCK 与 SIGIO 生成的 recvfrom 配对

Question

我正在使用 SIGIO 构建一个 UDP 非阻塞客户端和服务器套接字对，以促进在服务器上接收数据。这是我的情况。当客户向我发送消息时，我有一个 SIGIO 激活。但是，这些消息大约是。 3 个 1024 字节数据包突发（即客户端一次发送 3 个数据包）。我希望服务器在第一个数据包上触发 SIGIO，并且在处理程序中我有一个 while 循环，它最多遍历 3 个数据包中的 2 个并读取它们（对于当前运行）。但是，这3个数据包发送中有几个运行，它们之间相隔5秒。所以我希望第一个数据包触发 SIGIO，同时清空缓冲区，并且 在信号处理程序中 我取消了可能由剩余的 3-1 数据包引起的一个 SIGIO 未决信号在第一个运行上，并为下一个运行第一个数据包（在 5 秒内）重新启用 SIGIO。如果你只看代码会更容易解释：

客户：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <netdb.h>
#include <math.h>
#include <time.h>

static const unsigned int TIMEOUT_SECS = 2; // Seconds between retransmits

void error(const char *msg, const char *detail);

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

  if ((argc < 6) || (argc > 7))
    error("Parameter(s)", "<Server Address/Name> <Echo File> <Server Port/Service> <Maximum packet length> <Number of trials> [<Seconds between trials>]");

  char *server = argv[1];
  FILE *fp = fopen(argv[2], "rb");
  if (fp == NULL)
    error("Error opening file", "");

  fseek(fp, -1, SEEK_END);
  unsigned long stringlen = (getc(fp) == '\n' ? (ftell(fp) - 1) : ftell(fp));
  rewind(fp);

  char *string = malloc(stringlen * sizeof(char));
  if (string == NULL)
    error("malloc() failed", "");
  for (int i = 0; i < stringlen; i++) {
    *(string + i) = fgetc(fp);
  }
  printf("%lu\n", stringlen);

  fclose(fp);

  in_port_t service = (in_port_t) strtol(argv[3], NULL, 10);
  int maxpack = (int) strtol(argv[4], NULL, 10);
  int trials = (int) strtol(argv[5], NULL, 10); // must coordinate with server
  if (trials < 1)
    error("invalid trials parameter", "entered a nonpositive number");
  int sleeptime = (argc == 7) ? (int) strtol(argv[6], NULL, 10) : 5;

  int sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
  if (sock < 0)
    error("socket() failed", "");

  struct timeval timeout;
  timeout.tv_sec = TIMEOUT_SECS;
  timeout.tv_usec = 0;
  if (setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (char *) &timeout, sizeof(timeout)) < 0)
    error("setsockopt() failed", "");
  if (setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (char *) &timeout, sizeof(timeout)) < 0)
    error("setsockopt() failed", "");

  struct sockaddr_in servAddr;
  memset(&servAddr, 0, sizeof(servAddr));
  servAddr.sin_family = AF_INET;
  int rc = inet_pton(AF_INET, server, &servAddr.sin_addr.s_addr);
  if (rc == 0)
    error("inet_pton() failed", "invalid address string");
  else if (rc < 0)
    error("inet_pton() failed", "");
  servAddr.sin_port = htons(service);

  int packnum = (int) ceil((double) stringlen / maxpack);
  for (int num = 0; num < trials; num++) {
    struct timespec start, end;
    uint64_t diff;
    ssize_t numBytes;
    for (int i = 0; i < packnum; i++) {
      if (i == 0)
        clock_gettime(CLOCK_MONOTONIC, &start);
      numBytes = sendto(sock, string + i * maxpack, maxpack, 0, (struct sockaddr* ) &servAddr, sizeof(servAddr));
      if (numBytes < 0)
        error("sendto() failed", "");
    }

    struct sockaddr_in fromAddr;
    socklen_t fromAddrlen = sizeof(fromAddr);
    char buffer[strlen(argv[3]) + 1 + packnum + 1];
    memset(&buffer, 0, sizeof(buffer));
    do {
      numBytes = recvfrom(sock, buffer, sizeof(buffer) - 1, 0, (struct sockaddr *) &fromAddr, &fromAddrlen);
      if (numBytes < 0)
        error("recvfrom() failed", "");
      buffer[numBytes] = '[=11=]';
    } while ((int) strtol(buffer, NULL, 10) == num);

    clock_gettime(CLOCK_MONOTONIC, &end);
    diff = 1000000000L * (end.tv_sec - start.tv_sec) + end.tv_nsec - start.tv_nsec;

    printf("Received: %s\n", buffer);
    printf("Time: %f ms\n", (double) (diff / (double) 1000000));
    sleep(sleeptime);
  }

  free(string);
  close(sock);

  exit(0);
}

void error(const char *msg, const char *detail) {
  fputs(msg, stderr);
  if (*detail != '[=11=]') {
    fputs(": ", stderr);
    fputs(detail, stderr);
  }
  fputc('\n', stderr);
  exit(1);
}

服务器：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <limits.h>
#include <fcntl.h>
#include <signal.h>
#include <errno.h>

#define ITOA_BASE_N (sizeof(unsigned)*CHAR_BIT + 2)

void error(const char *msg, const char *detail);
void PrintSocketAddress(const struct sockaddr *address, FILE *stream);
void SIGIOHandler(int signalType);
char * itoa_base(char *s, int x, int base);
void recvflush(int sockfd);

#define ITOA(x,b) itoa_base((char [ITOA_BASE_N]){0} , (x), (b))

int sock, acknum, maxpack; // GLOBAL for signal handler
volatile sig_atomic_t run, packets; // VOLATILE FOR signal handler modification
struct sigaction handler;

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

  if (argc != 5)
    error("Parameter(s)", "<Server Port/Service> <Number of ACKs> <Maximum packet length> <Number of trials>");

  in_port_t service = (in_port_t) strtol(argv[1], NULL, 10);
  acknum = (int) strtol(argv[2], NULL, 10);
  maxpack = (int) strtol(argv[3], NULL, 10);
  int trials = (int) strtol(argv[4], NULL, 10); // must coordinate with client
  run = 0;

  sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
  if (sock < 0)
    error("socket() failed", strerror(errno));

  struct sockaddr_in servAddr;
  memset(&servAddr, 0, sizeof(servAddr));
  servAddr.sin_family = AF_INET;
  servAddr.sin_addr.s_addr = htonl(INADDR_ANY);
  servAddr.sin_port = htons(service);

  if (bind(sock, (struct sockaddr *) &servAddr, sizeof(servAddr)) < 0)
    error("bind() failed", strerror(errno));

  handler.sa_handler = SIGIOHandler;
  if (sigfillset(&handler.sa_mask) < 0)
    error("sigfillset() failed", "mask not set");
  handler.sa_flags = 0;

  if (sigaction(SIGIO, &handler, 0) < 0)
    error("sigaction() failed", "SIGIO behavior unable to be modified");

  if (fcntl(sock, F_SETOWN, getpid()) < 0)
    error("Unable to set process owner to us", "");

  if (fcntl(sock, F_SETFL, O_NONBLOCK | FASYNC) < 0)
    error(
        "Unable to put client sock into non-blocking/async mode", "");

  recvflush(sock);

  for (;;) {
    if (run == trials) {
      close(sock);
      exit(0);
    }
    printf(".");
    fflush(stdout);
    sleep(1);
  }
}

void SIGIOHandler(int signalType) {
  printf("SIGIO\n");
  packets = 0;
  run++;
  struct sockaddr_in clntAddr;
  socklen_t clntLen = sizeof(clntAddr);
  ssize_t numBytes;
  //FILE *fp = fopen("output.txt", "ab");
  //if (fp == NULL)
    //error("Error opening file", "");
  char buffer[maxpack + 1];
  //char *output = malloc(2000 * (maxpack + 1) * sizeof(char));
  //if (output == NULL)
    //error("malloc() failed", "");
  memset(buffer, 0, sizeof(buffer));
  //memset(output, 0, sizeof(output));
  while (packets < 2) { // read up to 2 packets (1024 bytes each) from buffer
    numBytes = recvfrom(sock, buffer, maxpack, 0, (struct sockaddr *) &clntAddr, &clntLen);
    if (numBytes < 0) {
      if (errno != EWOULDBLOCK) // only acceptable error
        error("recvfrom() failed", strerror(errno));
      else
        printf("%d EWOULDBLOCK", numBytes);
    } else {
      if (numBytes > 0) {
        packets++;
        buffer[numBytes] = '[=12=]';
        //memmove(output + maxpack * packets, buffer, strlen(buffer));
      }
      printf("Run-Packet numBytes: %d-%d %d\n", run, packets, numBytes);
      printf("\t%s\n", buffer);
    }
  }
  sigset_t mask;
  sigpending(&mask);
  if (sigismember(&mask, SIGIO))
    printf("pending SIGIO signal\n");
  printf("setting ignore\n");
  struct sigaction change;
  change.sa_handler = SIG_IGN; // set ignore handler to cancel pending SIGIO signals from 1024 - 1 packets
  if (sigfillset(&change.sa_mask) < 0)
    error("sigfillset() failed", "mask not set");
  change.sa_flags = 0;
  if (sigaction(SIGIO, &change, 0) < 0)
    error("sigaction() failed", "SIGIO behavior unable to be modified");
  sigpending(&mask);
  if (sigismember(&mask, SIGIO))
    printf("pending SIGIO signal\n"); // should NOT show up
  recvflush(sock); // flush ignored bytes
  //for (int i = 0; i < strlen(output); i++)
    //putc(output[i], fp);
  //putc('[=12=]', fp);

  //fclose(fp);
  //free(output);

  char *ack = ITOA(run, 10);
  char *append = ITOA(packets, 10);
  strcat(ack, "/"); // fix client side
  strcat(ack, append);
  for (int i = 0; i < acknum; i++) {
    numBytes = sendto(sock, ack, sizeof(ack), 0, (struct sockaddr *) &clntAddr, clntLen);
    if (numBytes < 0)
      error("sendto() failed", strerror(errno));
  }

  if (sigaction(SIGIO, &handler, 0) < 0)
        error("sigaction() failed", "SIGIO behavior unable to be modified"); // reenable SIGIO for subsequent runs

  printf("HERE\n");
}

char * itoa_base(char *s, int x, int base) {
  s += ITOA_BASE_N - 1;
  *s = '[=12=]';
  if (base >= 2 && base <= 36) {
    int x0 = x;
    do {
      *(--s) = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"[abs(x % base)];
      x /= base;
    } while (x);
    if (x0 < 0) {
      *(--s) = '-';
    }
  }
  return s;
}

void recvflush(int sockfd) {
  ssize_t num;
  do {
    num = recvfrom(sockfd, NULL, 0, 0, NULL, NULL);
    if (num < 0 && errno != EWOULDBLOCK)
      error("recvfrom() failed", strerror(errno));
  } while (errno != EWOULDBLOCK);
}

现在，我首先使用参数 8080 10 1024 2 启动服务器，然后使用 | head -c 10000 通过管道查看前 10,000 个字符。然后，我使用参数 192.168.4.101 test.txt 8080 1024 2 启动客户端。 test.txt 填充了大约。 3000个垃圾字符，应该分成3个1024个数据包（客户端注意ceil）。但是，每当我发送信息时，都会发生一件非常奇怪的事情。

有时，一切正常，我收到了预期的输出，缓冲区 printfs 之间有 5 秒的间隔。其他时候，它会搞砸并永远生成 EWOULDBLOCK 错误（请参阅服务器代码中的 printf("%d EWOULDBLOCK", numBytes);）。每次它生成此错误时，它仍然会读取被忽略的数据包（因为我只读取 3 个数据包中的 2 个并使用 recvflush 刷新其余数据包），即使我正在刷新接收缓冲区。据说，recvflush 会清除整个缓冲区以为后续的运行做准备，但由于某种原因，被忽略的数据包会保留在那里。求助！

谢谢

Answer 1

您假设在 SIGIO 触发您的信号处理程序时所有 3 个数据包都已经在套接字缓冲区中。但是 SIGIO 在第一个数据包上被触发，这意味着套接字缓冲区中可能只有一个或两个数据包。

假设只有一个数据包，第二个 recv 将因 EWOULDBLOCK 而失败，因为没有更多数据包。类似于 recvflush 中的 recv 将失败，因为丢失的数据包还不存在。如果数据包不存在，也不会触发 SIGIO，因此 sigpending 的任何调用都无法检测到有未完成的数据。

UDP 非阻塞 client/server 中生成的随机 EWOULDBLOCK 与 SIGIO 生成的 recvfrom 配对

Random EWOULDBLOCKs generated in UDP non-blocking client/server pair with SIGIO-generated recvfroms

c

sockets

networking

udp

asyncsocket