Multithreading serveur UDP avec epoll?

Question

Je voudrais développer un serveur UDP multithread dans C/Linux. Le service fonctionne sur un seul port x, il n'y a donc que la possibilité de lier un seul socket UDP. Afin de travailler sous des charges élevées, j'ai n threads (définis statiquement), disons 1 thread par CPU. Le travail peut être livré au thread en utilisant epoll_wait, ainsi les threads sont réveillés à la demande avec 'EPOLLET | EPOLLONESHOT '. J'ai joint un exemple de code:

static int epfd; 
static sig_atomic_t sigint = 0; 

... 

/* Thread routine with epoll_wait */ 
static void *process_clients(void *pevents) 
{ 
    int rc, i, sock, nfds; 
    struct epoll_event ep, *events = (struct epoll_event *) pevents; 

    while (!sigint) { 
     nfds = epoll_wait(epfd, events, MAX_EVENT_NUM, 500); 

     for (i = 0; i < nfds; ++i) { 
      if (events[i].data.fd < 0) 
       continue; 

      sock = events[i].data.fd; 

      if((events[i].events & EPOLLIN) == EPOLLIN) { 
       printf("Event dispatch!\n"); 
       handle_request(sock); // do a recvfrom 
      } else 
       whine("Unknown poll event!\n"); 

      memset(&ep, 0, sizeof(ep)); 
      ep.events = EPOLLIN | EPOLLET | EPOLLONESHOT; 
      ep.data.fd = sock; 

      rc = epoll_ctl(epfd, EPOLL_CTL_MOD, sock, &ep); 
      if(rc < 0) 
       error_and_die(EXIT_FAILURE, "Cannot add socket to epoll!\n"); 
     } 
    } 

    pthread_exit(NULL); 
} 

int main(int argc, char **argv) 
{ 
    int rc, i, cpu, sock, opts; 
    struct sockaddr_in sin; 
    struct epoll_event ep, *events; 
    char *local_addr = "192.168.1.108"; 
    void *status; 
    pthread_t *threads = NULL; 
    cpu_set_t cpuset; 

    threads = xzmalloc(sizeof(*threads) * MAX_THRD_NUM); 
    events = xzmalloc(sizeof(*events) * MAX_EVENT_NUM); 

    sock = socket(PF_INET, SOCK_DGRAM, 0); 
    if (sock < 0) 
     error_and_die(EXIT_FAILURE, "Cannot create socket!\n"); 

    /* Non-blocking */ 
    opts = fcntl(sock, F_GETFL); 
    if(opts < 0) 
     error_and_die(EXIT_FAILURE, "Cannot fetch sock opts!\n"); 
    opts |= O_NONBLOCK; 
    rc = fcntl(sock, F_SETFL, opts); 
    if(rc < 0) 
     error_and_die(EXIT_FAILURE, "Cannot set sock opts!\n"); 

    /* Initial epoll setup */ 
    epfd = epoll_create(MAX_EVENT_NUM); 
    if(epfd < 0) 
     error_and_die(EXIT_FAILURE, "Error fetching an epoll descriptor!\n"); 

    memset(&ep, 0, sizeof(ep)); 
    ep.events = EPOLLIN | EPOLLET | EPOLLONESHOT; 
    ep.data.fd = sock; 

    rc = epoll_ctl(epfd, EPOLL_CTL_ADD, sock, &ep); 
    if(rc < 0) 
     error_and_die(EXIT_FAILURE, "Cannot add socket to epoll!\n"); 

    /* Socket binding */ 
    sin.sin_family = AF_INET; 
    sin.sin_addr.s_addr = inet_addr(local_addr); 
    sin.sin_port = htons(port_xy); 

    rc = bind(sock, (struct sockaddr *) &sin, sizeof(sin)); 
    if (rc < 0) 
     error_and_die(EXIT_FAILURE, "Problem binding to port! " 
         "Already in use?\n"); 

    register_signal(SIGINT, &signal_handler); 

    /* Thread initialization */ 
    for (i = 0, cpu = 0; i < MAX_THRD_NUM; ++i) { 
     rc = pthread_create(&threads[i], NULL, process_clients, events); 
     if (rc != 0) 
      error_and_die(EXIT_FAILURE, "Cannot create pthread!\n"); 

     CPU_ZERO(&cpuset); 
     CPU_SET(cpu, &cpuset); 

     rc = pthread_setaffinity_np(threads[i], sizeof(cpuset), &cpuset); 
     if (rc != 0) 
      error_and_die(EXIT_FAILURE, "Cannot create pthread!\n"); 

     cpu = (cpu + 1) % NR_CPUS_ON; 
    } 

    printf("up and running!\n"); 

    /* Thread joining */ 
    for (i = 0; i < MAX_THRD_NUM; ++i) { 
     rc = pthread_join(threads[i], &status); 
     if (rc != 0) 
      error_and_die(EXIT_FAILURE, "Error on thread exit!\n"); 
    } 

    close(sock); 
    xfree(threads); 
    xfree(events); 

    printf("shut down!\n"); 

    return 0; 
} 

Est-ce la bonne façon de gérer ce scénario avec epoll? Est-ce que la fonction _handle_request_ devrait retourner aussi vite que possible, car pour l'instant la file d'attente pour le socket est bloquée ?!

Merci pour vos réponses!

Answer 1

Comme vous n'utilisez qu'un seul socket UDP, cela ne sert à rien d'utiliser epoll - utilisez plutôt un recvfrom bloquant. Maintenant, en fonction du protocole que vous devez gérer - si vous pouvez traiter chaque paquet UDP individuellement - vous pouvez appeler recvfrom simultanément à partir de plusieurs threads (dans un pool de threads). Le système d'exploitation veillera à ce qu'un seul thread reçoive le paquet UDP. Ce thread peut alors faire tout ce qu'il doit faire dans handle_request.

Cependant, si vous avez besoin pour traiter les paquets UDP dans un ordre particulier, vous avez probablement pas que de nombreuses occasions de parallalise votre programme ...

Answer 2

Non, cela ne fonctionnera pas comme vous le souhaitez. Pour que les threads de travail traitent les événements arrivant via une interface epoll, vous avez besoin d'une architecture différente.

design exemple (il y a plusieurs façons de le faire) Utilisations: sémaphores SysV/POSIX.

• Demandez le spawn de fil maître n subthreads et sémaphores, puis bloquer epolling vos prises de courant (ou autre).

• Avoir chaque sous-bloc sur le sémaphore.

• Lorsque le thread principal se débloque, il stocke les événements dans une structure globale et augmente le sémaphore une fois par événement.

• Les subthreads débloquent, traiter les événements, le bloc à nouveau lorsque les rendements sémaphores à 0.

Vous pouvez utiliser un tuyau partagé entre les threads pour obtenir des fonctionnalités très similaires à celle de la sémaphores. Cela vous permet de bloquer select() au lieu du sémaphore, que vous pouvez utiliser pour réveiller les threads sur un autre événement (délais, autres canaux, etc.)

Vous pouvez également inverser ce contrôle et avoir le thread principal se réveiller lorsque ses travailleurs exigent des tâches. Je pense que l'approche ci-dessus est préférable pour votre cas, cependant.