Wielowątkowość serwera UDP z epoll?

Question

Chciałbym stworzyć wielowątkowy serwer UDP w C/Linux. Usługa działa na jednym porcie x, dlatego istnieje tylko możliwość powiązania z nim jednego gniazda UDP. Aby pracować pod dużym obciążeniem, mam n wątków (statycznie zdefiniowanych), powiedzmy 1 wątek na procesor. Praca mogła zostać dostarczona do wątku za pomocą epoll_wait, więc wątki są budzone na żądanie za pomocą "EPOLLET | EPOLLONESHOT ". Dołączyłem przykład kodu:

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; 
} 

Czy to właściwy sposób postępowania z tym scenariuszem z epoll? Czy funkcja _handle_request_ powinna powrócić tak szybko, jak to możliwe, ponieważ na tę chwilę sekwencja zdarzeń dla gniazda jest zablokowana ?!

Dzięki za odpowiedzi!

Answer 1

Ponieważ używasz tylko jednego gniazda UDP, nie ma sensu używać epollu - zamiast tego użyj blokującego recvfrom.

Teraz, w zależności od protokołu, który musisz obsługiwać - jeśli możesz przetwarzać każdy pakiet UDP osobno - możesz wywoływać recvfrom jednocześnie z wielu wątków (w puli wątków). System operacyjny zadba, aby dokładnie jeden wątek otrzymał pakiet UDP. Wątek ten może następnie zrobić wszystko, co musi zrobić w handle_request.

Jednak jeśli trzeba przetwarzać pakiety UDP w określonej kolejności, prawdopodobnie nie ma to wiele możliwości parallalise program ...

Answer 2

Nie, to nie będzie działać tak, jak chcesz. Aby wątki robocze przetwarzały zdarzenia przychodzące przez interfejs epoli, potrzebna jest inna architektura.

Przykład projektowania (istnieje kilka sposobów, aby to zrobić) Zastosowanie: semafory SysV/POSIX.

• Czy wątku ikry n subthreads master i semafora, a następnie zablokować epolling swoich gniazd (lub cokolwiek).

• Każdemu blokowi podprogramu na spadek semafora.

• Gdy główna nitka odblokowuje się, przechowuje zdarzenia w jakiejś globalnej strukturze i podnosi semafor raz na zdarzenie.

• W subthreads odblokować, przetworzenia zdarzeń blok ponownie, gdy semafor powraca do 0.

Można użyć rury wspólny wśród wszystkich wątków w celu osiągnięcia bardzo podobną funkcjonalność do semafora. Umożliwiłoby to zablokowanie select() zamiast semafora, którego można użyć do wybudzenia wątków w innym zdarzeniu (limity czasu, inne przewody itp.).

Można również odwrócić tę kontrolę i uzyskać główną ścieżkę obudź się, gdy jego pracownicy żądają zadań. Myślę, że powyższe podejście jest lepsze dla twojej sprawy.