epoll惊群学习

#include stdio.h 
#include sys/types.h 
#include sys/socket.h 
#include unistd.h 
#include sys/epoll.h 
#include netdb.h 
#include stdlib.h 
#include fcntl.h 
#include sys/wait.h 
#include errno.h 
#define PROCESS_NUM 4
#define MAXEVENTS 64

#define MAXLINE 80
#define SERV_PORT 9999
#define OPEN_MAX 526

static int pidnum[PROCESS_NUM],epfd[PROCESS_NUM];
void create(int listenfd,int num);
typedef struct {
    int fd[2];
} eptest;
int main(int argc, char *argv[])
{
    int i,r;
    eptest epdata[5];
    struct sockaddr_in servaddr,cliaddr;
    socklen_t cliaddr_len;
    int listenfd,connfd,sockfd;
    int nready,efd,res,client[OPEN_MAX];
    ssize_t n;
    char buf[MAXLINE];
    char str[INET_ADDRSTRLEN];
    
    listenfd = socket(AF_INET, SOCK_STREAM, 0);
    printf("listenfd:%d\n",listenfd);


    bzero(&servaddr, sizeof(servaddr));
    //ipv4 
    servaddr.sin_family= AF_INET;
    //ip 
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    //port
    servaddr.sin_port= htons(SERV_PORT);

    bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr));
    //设置同时能够接受128个请求，注意是同一时间
    listen(listenfd, 128);
    printf("pid = %d\n",getpid());
    for(i = 0;i < PROCESS_NUM;i++){
    r = socketpair(AF_UNIX, SOCK_STREAM, 0, epdata[i].fd);
    if (r < 0){
        perror( "socketpair()" );
        exit(1);
    }
    create(listenfd,i);
    }
    printf("pid1 = %d i = %d\n",getpid(),i);
    while(1);
 
    return 0;
}

void create(int listenfd,int num){
  int epoll_fd,pid;
  struct epoll_event event;
  struct epoll_event events;
  struct epoll_event tep,ep;

  pid = fork();
  /*switch (pid) {
    case -1:
        return -1;
    case 0:
        break;
    default:
        return 0;
  }*/
 if(pid == 0){ 
  pidnum[num] = getpid();  

  if((epoll_fd = epoll_create(MAXEVENTS))< 0){
        perror("epoll_create");
        exit(1);
  }
  event.data.fd = listenfd;
  event.events = EPOLLIN | EPOLLEXCLUSIVE;
  if(epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listenfd, &event) < 0){
        perror("epoll_ctl");
        exit(1);
  }
  printf("pidnum[%d] = %d pid = %d epoll_fd = %d\n",num,pidnum[num],getpid(),epoll_fd); 
  while(1){
                int num, j,i;
                num = epoll_wait(epoll_fd, &events, MAXEVENTS, -1);
                printf("process %d returnt from epoll_wait epoll_fd = %d\n", getpid(),epoll_fd);
sleep(2);
                    if((events.events & EPOLLERR) || (events.events & EPOLLHUP) || (!(events.events & EPOLLIN))){
                        fprintf(stderr, "epoll error\n");
                        close(events.data.fd);
                        continue;
                    }else if(listenfd == events.data.fd){
                        //收到关于监听套接字的通知，意味着一盒或者多个传入连接
                        struct sockaddr in_addr;
                        socklen_t in_len = sizeof(in_addr);
                        if(accept(listenfd, &in_addr, &in_len) < 0){
                            printf("process %d accept failed!\n", getpid());
                        }else{
                            printf("process %d accept successful!\n", getpid());
                        }
                    }
                
 }
}
}

----------------------------------linux-5.4.1
/*
 * Must be called with "mtx" held.
 */
static int ep_insert(struct eventpoll *ep, const struct epoll_event *event,
     struct file *tfile, int fd, int full_check)
{
int error, pwake = 0;
__poll_t revents;
long user_watches;
struct epitem *epi;
struct ep_pqueue epq;

lockdep_assert_irqs_enabled();

user_watches = atomic_long_read(&ep->user->epoll_watches);
if (unlikely(user_watches >= max_user_watches))
return -ENOSPC;
if (!(epi = kmem_cache_alloc(epi_cache, GFP_KERNEL)))
return -ENOMEM;

/* Item initialization follow here ... */
INIT_LIST_HEAD(&epi->rdllink);
INIT_LIST_HEAD(&epi->fllink);
INIT_LIST_HEAD(&epi->pwqlist);
epi->ep = ep;
ep_set_ffd(&epi->ffd, tfile, fd);
epi->event = *event;
epi->nwait = 0;
epi->next = EP_UNACTIVE_PTR;
if (epi->event.events & EPOLLWAKEUP) {
error = ep_create_wakeup_source(epi);
if (error)
goto error_create_wakeup_source;
} else {
RCU_INIT_POINTER(epi->ws, NULL);
}

/* Initialize the poll table using the queue callback */
epq.epi = epi;
/*
/*
 * This is the callback that is used to add our wait queue to the
 * target file wakeup lists.
 */
static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,
poll_table *pt)
{
struct epitem *epi = ep_item_from_epqueue(pt);
struct eppoll_entry *pwq;

if (epi->nwait >= 0 && (pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL))) {
init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
pwq->whead = whead;
pwq->base = epi;
if (epi->event.events & EPOLLEXCLUSIVE)
add_wait_queue_exclusive(whead, &pwq->wait);
else
add_wait_queue(whead, &pwq->wait);
list_add_tail(&pwq->llink, &epi->pwqlist);
epi->nwait++;
} else {
/* We have to signal that an error occurred */
epi->nwait = -1;
}
}
*/
init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);
}