İş parçacığıyla aynı epoll dosya tanıtıcısını paylaşmak tamam mı?

Question

Birkaç iş parçacığıyla aynı Epoll fd (soket değil) paylaşmayı güvenli mi? Ve eğer öyleyse, her bir iş parçacığı kendi olay dizisini epoll_wait(2)'a geçirmek zorunda mı yoksa paylaşır mı?

void *thread_func(void *socket_fd) { 
     // create events_fd 
     // allocate memory for events array 
     // subscribe to events EPOLLIN and EPOLLET 
     // epoll_wait using own epoll_fd and events_array 
     // now all threads would have a separate epoll_fd with 
     // events populated on its own array 
    } 

    void main(void) { 
    // create and bind to socket 

    //create multiple threads and pass thread_func and socket_fd to 
    // all threads 
    } 

C bunun nasıl iyi bir örnek var mı: Örneğin

void *thread_func(void *thread_args) { 
     // extract socket_fd, epoll_fd, &event, &events_array from 
     //  thread_args 
     // epoll_wait() using epoll_fd and events_array received from main 
     // now all threads would be using same epoll_fd and events array 
    } 

    void main(void) { 
     // create and bind to socket 
     // create events_fd 
     // allocate memory for events array 
     // subscribe to events EPOLLIN and EPOLLET 
     // pack the socket_fd, epoll_fd, &events, &events_array into 
     // thread_args struct. 

     // create multiple threads and pass thread_func and 
     // same thread_args to all threads 
    } 

Yoksa böyle yapmak daha iyidir İçin

? Gördüğüm örnekler, olay döngüsünü main()'da çalıştırır ve bir olay algılandığında isteği işlemek için yeni bir iş parçacığı oluşturur. Yapmak istediğim programın başında belirli bir sayıda iş parçacığı oluşturmak ve her iş parçacığı olay döngüsünü ve işlem isteklerini çalıştırıyor.

Answer 1

Aynı Epoll fd'yi (soket değil) birkaç iş parçacığı arasında paylaşmanın güvenli midir?

Evet, güvenlidir - epoll(7) arayüz evreli - eğer en azından kullanmak gerekir, ancak bunu yaparken dikkatli olmak gerekir EPOLLET (kenar tetiklemeli modu, varsayılan aksine seviye tetiklemeli) Diğer konularda sahte uyandırmadan kaçınmak için. Bunun nedeni, seviye tetiklemeli modun, işlenmek üzere yeni bir etkinlik olduğunda her iş parçacığı tarafından uyanmasıdır. Sadece bir iş parçacığı bununla uğraşacağından, çoğu ileti gereksiz yere uyanırdı. Paylaşılan epfd her iş parçacığının kendi olaylar dizi veya paylaşılan olaylar dizisi) (epoll_wait geçmek zorunda kalacak kullanılırsa

Evet, her iş parçacığı üzerinde ayrı olaylar dizisini ihtiyaç veya başka yarış koşullarına sahip olacaksınız ve kötü şeyler olabilir. Örneğin, epoll_wait(2) tarafından döndürülen olaylar arasında yinelenen ve aniden başka bir iş parçacığının aynı dizi ile epoll_wait(2) numaralı çağrıları çağırdığında ve ardından diğer iş parçacığı bunları okurken olayların üzerine yazıldığında istekleri işleyen bir iş parçacığınız olabilir. İyi değil! kesinlikle her bir iş parçacığı için ayrı bir diziye ihtiyacınız var.

Her bir iş parçacığı için ayrı bir diziniz varsayarsak, olasılık - aynı epoll fd üzerinde beklemek veya her iş parçacığı için ayrı bir epoll'a sahip olmak - eşit olarak iyi çalışacaktır, ancak semantiklerin farklı olduğunu unutmayın. Genel olarak paylaşılan bir epoll fd ile, her iş parçacığı istemcisinden bir istek bekler, çünkü istemcilerin hepsi aynı epoll fd'ye eklenir. Her iş parçacığı için ayrı bir epoll fd ile, her iş parçacığı, istemcinin bir alt kümesinden (bu iş parçacığı tarafından kabul edilen istemciler) sorumludur.

Bu, sisteminiz için ilgisiz olabilir veya büyük bir fark yaratabilir. Örneğin, bir iş parçacığının, bir grup güç kullanıcısını ağır ve sık talepte bulunan bir grup elde etmek için talihsiz olması, bu iş parçacığını fazla çalışmaması, daha az saldırgan istemcileri olan diğer iş parçacıklarının neredeyse boşta kalması da talihsiz bir durum olabilir. Bu haksızlık olmaz mıydı? Öte yandan, belki de belirli bir kullanıcı sınıfı ile uğraşan sadece bazı konulara sahip olmak isteyebilirsiniz ve bu durumda belki her iş parçacığında farklı epoll fds olması mantıklıdır. Her zamanki gibi, her iki olasılığı da göz önünde bulundurmanız, ticaret fırsatlarını değerlendirmeniz, özel probleminizi düşünmeniz ve bir karar vermeniz gerekiyor.

Aşağıda global olarak paylaşılan bir epoll fd'nin kullanıldığı bir örnektir.Aslında bunu yapmayı planlamamıştım, ama bir şey başka bir şeye yol açtı ve iyi eğlenceliydi ve başlamanıza yardımcı olabileceğini düşünüyorum. Port 3000'ü dinleyen ve aynı anda yeni istemcileri kabul etmek ve istekleri sunmak için epoll'u kullanan 20 iş parçacığı havuzu olan bir eko sunucusu.

#include <stdio.h> 
#include <stdlib.h> 
#include <inttypes.h> 
#include <errno.h> 
#include <string.h> 
#include <pthread.h> 
#include <assert.h> 
#include <unistd.h> 
#include <sys/types.h> 
#include <sys/socket.h> 
#include <arpa/inet.h> 
#include <sys/epoll.h> 

#define SERVERPORT 3000 
#define SERVERBACKLOG 10 
#define THREADSNO 20 
#define EVENTS_BUFF_SZ 256 

static int serversock; 
static int epoll_fd; 
static pthread_t threads[THREADSNO]; 

int accept_new_client(void) { 

    int clientsock; 
    struct sockaddr_in addr; 
    socklen_t addrlen = sizeof(addr); 
    if ((clientsock = accept(serversock, (struct sockaddr *) &addr, &addrlen)) < 0) { 
     return -1; 
    } 

    char ip_buff[INET_ADDRSTRLEN+1]; 
    if (inet_ntop(AF_INET, &addr.sin_addr, ip_buff, sizeof(ip_buff)) == NULL) { 
     close(clientsock); 
     return -1; 
    } 

    printf("*** [%p] Client connected from %s:%" PRIu16 "\n", (void *) pthread_self(), 
      ip_buff, ntohs(addr.sin_port)); 

    struct epoll_event epevent; 
    epevent.events = EPOLLIN | EPOLLET; 
    epevent.data.fd = clientsock; 

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, clientsock, &epevent) < 0) { 
     perror("epoll_ctl(2) failed attempting to add new client"); 
     close(clientsock); 
     return -1; 
    } 

    return 0; 
} 

int handle_request(int clientfd) { 
    char readbuff[512]; 
    struct sockaddr_in addr; 
    socklen_t addrlen = sizeof(addr); 
    ssize_t n; 

    if ((n = recv(clientfd, readbuff, sizeof(readbuff)-1, 0)) < 0) { 
     return -1; 
    } 

    if (n == 0) { 
     return 0; 
    } 

    readbuff[n] = '\0'; 

    if (getpeername(clientfd, (struct sockaddr *) &addr, &addrlen) < 0) { 
     return -1; 
    } 

    char ip_buff[INET_ADDRSTRLEN+1]; 
    if (inet_ntop(AF_INET, &addr.sin_addr, ip_buff, sizeof(ip_buff)) == NULL) { 
     return -1; 
    } 

    printf("*** [%p] [%s:%" PRIu16 "] -> server: %s", (void *) pthread_self(), 
      ip_buff, ntohs(addr.sin_port), readbuff); 

    ssize_t sent; 
    if ((sent = send(clientfd, readbuff, n, 0)) < 0) { 
     return -1; 
    } 

    readbuff[sent] = '\0'; 

    printf("*** [%p] server -> [%s:%" PRIu16 "]: %s", (void *) pthread_self(), 
      ip_buff, ntohs(addr.sin_port), readbuff); 

    return 0; 
} 

void *worker_thr(void *args) { 
    struct epoll_event *events = malloc(sizeof(*events)*EVENTS_BUFF_SZ); 
    if (events == NULL) { 
     perror("malloc(3) failed when attempting to allocate events buffer"); 
     pthread_exit(NULL); 
    } 

    int events_cnt; 
    while ((events_cnt = epoll_wait(epoll_fd, events, EVENTS_BUFF_SZ, -1)) > 0) { 
     int i; 
     for (i = 0; i < events_cnt; i++) { 
      assert(events[i].events & EPOLLIN); 

      if (events[i].data.fd == serversock) { 
       if (accept_new_client() == -1) { 
        fprintf(stderr, "Error accepting new client: %s\n", 
         strerror(errno)); 
       } 
      } else { 
       if (handle_request(events[i].data.fd) == -1) { 
        fprintf(stderr, "Error handling request: %s\n", 
         strerror(errno)); 
       } 
      } 
     } 
    } 

    if (events_cnt == 0) { 
     fprintf(stderr, "epoll_wait(2) returned 0, but timeout was not specified...?"); 
    } else { 
     perror("epoll_wait(2) error"); 
    } 

    free(events); 

    return NULL; 
} 

int main(void) { 
    if ((serversock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) { 
     perror("socket(2) failed"); 
     exit(EXIT_FAILURE); 
    } 

    struct sockaddr_in serveraddr; 
    serveraddr.sin_family = AF_INET; 
    serveraddr.sin_port = htons(SERVERPORT); 
    serveraddr.sin_addr.s_addr = INADDR_ANY; 

    if (bind(serversock, (const struct sockaddr *) &serveraddr, sizeof(serveraddr)) < 0) { 
     perror("bind(2) failed"); 
     exit(EXIT_FAILURE); 
    } 

    if (listen(serversock, SERVERBACKLOG) < 0) { 
     perror("listen(2) failed"); 
     exit(EXIT_FAILURE); 
    } 

    if ((epoll_fd = epoll_create(1)) < 0) { 
     perror("epoll_create(2) failed"); 
     exit(EXIT_FAILURE); 
    } 

    struct epoll_event epevent; 
    epevent.events = EPOLLIN | EPOLLET; 
    epevent.data.fd = serversock; 

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, serversock, &epevent) < 0) { 
     perror("epoll_ctl(2) failed on main server socket"); 
     exit(EXIT_FAILURE); 
    } 

    int i; 
    for (i = 0; i < THREADSNO; i++) { 
     if (pthread_create(&threads[i], NULL, worker_thr, NULL) < 0) { 
      perror("pthread_create(3) failed"); 
      exit(EXIT_FAILURE); 
     } 
    } 

    /* main thread also contributes as worker thread */ 
    worker_thr(NULL); 

    return 0; 
} 

birkaç not: (Eğer örnekte gösterdiği gibi)

• main()int değil void dönmelidir
• daima hata dönüş kodları ile uğraşmak. Onları görmezden gelmek çok yaygındır ve işler kırıldığında ne olduğunu bilmek zordur.
• Kod, hiçbir isteğin 511 bayttan daha büyük olmadığını varsayar (handle_request() numaralı arabellek boyutunda görüldüğü gibi). Bir istek bundan daha büyükse, soket içinde çok uzun bir süre kalmış olabilir, çünkü epoll_wait(2) bu dosya tanıtıcısında yeni bir olay ortaya çıkana kadar bunu bildirmez (çünkü EPOLLET kullanıyoruz). En kötü durumda, müşteri asla yeni bir veri göndermeyebilir ve sonsuza kadar bir cevap bekleyebilir.
• Her istek için iş parçacığı tanımlayıcısını basan kod, pthread_t'un opak bir işaretçi türü olduğunu varsayar. Gerçekten de, pthread_t Linux'ta bir işaretçi türüdür, ancak diğer platformlarda bir tamsayı türü olabilir, bu nedenle bu taşınabilir değildir. Ancak, epoll Linux'a özel olduğundan, bu muhtemelen bir problem değildir, bu nedenle kod zaten taşınabilir değildir.
• Bir iş parçacığı hala bu istemciden istekte bulunduğunda, aynı istemciden başka istek gelmediğini varsayar. Bu arada yeni bir talep geldiğinde ve başka bir iş parçacığı bunu sunmaya başlarsa, bir yarış durumumuz var ve istemci eko mesajlarını gönderdiği aynı sırayla almaz (ancak, write(2) atomiktir, bu nedenle yanıtlar olabilirken sırasız, kesişmezler).