/*
 *  epoll_server()
 *	wait on connections and read data
 */
static void epoll_server(
	const int child,
	uint64_t *const counter,
	const uint32_t instance,
	const uint64_t max_ops,
	const char *name,
	const pid_t ppid)
{
	int efd = -1, sfd = -1, rc = EXIT_SUCCESS;
	int so_reuseaddr = 1;
	int port = opt_epoll_port + child + (max_servers * instance);
	struct sigaction new_action;
	struct epoll_event *events = NULL;
	struct sockaddr *addr = NULL;
	socklen_t addr_len = 0;

	new_action.sa_handler = handle_socket_sigalrm;
	sigemptyset(&new_action.sa_mask);
	new_action.sa_flags = 0;
	if (sigaction(SIGALRM, &new_action, NULL) < 0) {
		pr_fail_err(name, "sigaction");
		rc = EXIT_FAILURE;
		goto die;
	}
	if ((sfd = socket(opt_epoll_domain, SOCK_STREAM, 0)) < 0) {
		pr_fail_err(name, "socket");
		rc = EXIT_FAILURE;
		goto die;
	}
	if (setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &so_reuseaddr, sizeof(so_reuseaddr)) < 0) {
		pr_fail_err(name, "setsockopt");
		rc = EXIT_FAILURE;
		goto die_close;
	}

	stress_set_sockaddr(name, instance, ppid,
		opt_epoll_domain, port, &addr, &addr_len);

	if (bind(sfd, addr, addr_len) < 0) {
		pr_fail_err(name, "bind");
		rc = EXIT_FAILURE;
		goto die_close;
	}
	if (epoll_set_fd_nonblock(sfd) < 0) {
		pr_fail_err(name, "setting socket to non-blocking");
		rc = EXIT_FAILURE;
		goto die_close;
	}
	if (listen(sfd, SOMAXCONN) < 0) {
		pr_fail_err(name, "listen");
		rc = EXIT_FAILURE;
		goto die_close;
	}
	if ((efd = epoll_create1(0)) < 0) {
		pr_fail_err(name, "epoll_create1");
		rc = EXIT_FAILURE;
		goto die_close;
	}
	if (epoll_ctl_add(efd, sfd) < 0) {
		pr_fail_err(name, "epoll ctl add");
		rc = EXIT_FAILURE;
		goto die_close;
	}
	if ((events = calloc(MAX_EPOLL_EVENTS, sizeof(struct epoll_event))) == NULL) {
		pr_fail_err(name, "epoll ctl add");
		rc = EXIT_FAILURE;
		goto die_close;
	}

	do {
		int n, i;

		memset(events, 0, MAX_EPOLL_EVENTS * sizeof(struct epoll_event));
		errno = 0;

		/*
		 * Wait for 100ms for an event, allowing us to
		 * to break out if opt_do_run has been changed
		 */
		n = epoll_wait(efd, events, MAX_EPOLL_EVENTS, 100);
		if (n < 0) {
			if (errno != EINTR) {
				pr_fail_err(name, "epoll_wait");
				rc = EXIT_FAILURE;
				goto die_close;
			}
			break;
		}

		for (i = 0; i < n; i++) {
			if ((events[i].events & EPOLLERR) ||
			    (events[i].events & EPOLLHUP) ||
			    (!(events[i].events & EPOLLIN))) {
				/*
				 *  Error has occurred or fd is not
				 *  for reading anymore.. so reap fd
//.........这里部分代码省略.........

int
main (int argc, char *argv[])
{
    int sfd, s;
    int efd;
    struct epoll_event event;
    struct epoll_event *events;
    
    if (argc != 2)
    {
        fprintf (stderr, "Usage: %s [port]\n", argv[0]);
        exit (EXIT_FAILURE);
    }
    
    sfd = create_and_bind (argv[1]);
    if (sfd == -1)
        abort ();
    
    s = make_socket_non_blocking (sfd);
    if (s == -1)
        abort ();
    
    s = listen (sfd, SOMAXCONN);
    if (s == -1)
    {
        perror ("listen");
        abort ();
    }
    
    efd = epoll_create1 (0);
    if (efd == -1)
    {
        perror ("epoll_create");
        abort ();
    }
    
    event.data.fd = sfd;
    event.events = EPOLLIN | EPOLLET;
    s = epoll_ctl (efd, EPOLL_CTL_ADD, sfd, &event);
    if (s == -1)
    {
        perror ("epoll_ctl");
        abort ();
    }
    
    /* Buffer where events are returned */
    events = calloc (MAXEVENTS, sizeof event);
    
    /* The event loop */
    while (1)
    {
        int n, i;
        
        n = epoll_wait (efd, events, MAXEVENTS, -1);
        for (i = 0; i < n; i++)
        {
            if ((events[i].events & EPOLLERR) ||
                (events[i].events & EPOLLHUP) ||
                (!(events[i].events & EPOLLIN)))
            {
                /* An error has occured on this fd, or the socket is not
                 ready for reading (why were we notified then?) */
                fprintf (stderr, "epoll error\n");
                close (events[i].data.fd);
                continue;
            }
            
            else if (sfd == events[i].data.fd)
            {
                /* We have a notification on the listening socket, which
                 means one or more incoming connections. */
                while (1)
                {
                    struct sockaddr in_addr;
                    socklen_t in_len;
                    int infd;
                    char hbuf[NI_MAXHOST], sbuf[NI_MAXSERV];
                    
                    in_len = sizeof in_addr;
                    infd = accept (sfd, &in_addr, &in_len);
                    if (infd == -1)
                    {
                        if ((errno == EAGAIN) ||
                            (errno == EWOULDBLOCK))
                        {
                            /* We have processed all incoming
                             connections. */
                            break;
                        }
                        else
                        {
                            perror ("accept");
                            break;
                        }
                    }
                    
                    s = getnameinfo (&in_addr, in_len,
                                     hbuf, sizeof hbuf,
                                     sbuf, sizeof sbuf,
                                     NI_NUMERICHOST | NI_NUMERICSERV);
//.........这里部分代码省略.........

int fi_ibv_eq_open(struct fid_fabric *fabric, struct fi_eq_attr *attr,
		   struct fid_eq **eq, void *context)
{
	struct fi_ibv_eq *_eq;
	struct epoll_event event;
	int ret;

	_eq = calloc(1, sizeof *_eq);
	if (!_eq)
		return -ENOMEM;

	_eq->fab = container_of(fabric, struct fi_ibv_fabric,
				util_fabric.fabric_fid);

	fastlock_init(&_eq->lock);
	ret = dlistfd_head_init(&_eq->list_head);
	if (ret) {
		VERBS_INFO(FI_LOG_EQ, "Unable to initialize dlistfd\n");
		goto err1;
	}

	_eq->epfd = epoll_create1(0);
	if (_eq->epfd < 0) {
		ret = -errno;
		goto err2;
	}

	memset(&event, 0, sizeof(event));
	event.events = EPOLLIN;

	if (epoll_ctl(_eq->epfd, EPOLL_CTL_ADD,
		      _eq->list_head.signal.fd[FI_READ_FD], &event)) {
		ret = -errno;
		goto err3;
	}

	switch (attr->wait_obj) {
	case FI_WAIT_NONE:
	case FI_WAIT_UNSPEC:
	case FI_WAIT_FD:
		_eq->channel = rdma_create_event_channel();
		if (!_eq->channel) {
			ret = -errno;
			goto err3;
		}

		ret = fi_fd_nonblock(_eq->channel->fd);
		if (ret)
			goto err4;

		if (epoll_ctl(_eq->epfd, EPOLL_CTL_ADD, _eq->channel->fd, &event)) {
			ret = -errno;
			goto err4;
		}

		break;
	default:
		ret = -FI_ENOSYS;
		goto err1;
	}

	_eq->flags = attr->flags;
	_eq->eq_fid.fid.fclass = FI_CLASS_EQ;
	_eq->eq_fid.fid.context = context;
	_eq->eq_fid.fid.ops = &fi_ibv_eq_fi_ops;
	_eq->eq_fid.ops = &fi_ibv_eq_ops;

	*eq = &_eq->eq_fid;
	return 0;
err4:
	if (_eq->channel)
		rdma_destroy_event_channel(_eq->channel);
err3:
	close(_eq->epfd);
err2:
	dlistfd_head_free(&_eq->list_head);
err1:
	fastlock_destroy(&_eq->lock);
	free(_eq);
	return ret;
}

int
main(int argc, char **argv) {
    int n, nfds, res;
    struct itimerspec timerits;
    struct epoll_event events[MAX_EVENTS];
    struct epoll_event timerevent;
    IxpClient* client;
    struct sb sb;

    signals_setup(&quit_handler);

    struct sb_entry sbe_sda = {
        .sbe_path = "/rbar/60_sda",
        .sbe_private = "sda",
        .sbe_init = &init_block,
        .sbe_update = &update_block,
        .sbe_foreground = 0xbbbbbb,
        .sbe_background = 0x444444,
        .sbe_border     = 0x555555,        
    };

    struct sb_entry sbe_sdb = {
        .sbe_path = "/rbar/61_sdb",
        .sbe_private = "sdb",
        .sbe_init = &init_block,
        .sbe_update = &update_block,
        .sbe_foreground = 0xbbbbbb,
        .sbe_background = 0x444444,
        .sbe_border     = 0x555555,        
    };

    struct sb_entry sbe_sdc = {
        .sbe_path = "/rbar/62_sdc",
        .sbe_private = "sdc",
        .sbe_init = &init_block,
        .sbe_update = &update_block,
        .sbe_foreground = 0xbbbbbb,
        .sbe_background = 0x444444,
        .sbe_border     = 0x555555,        
    };

    int epollfd = epoll_create1(EPOLL_CLOEXEC);
    if(epollfd == -1) {
        perror("epoll_create");
        abort();
    }

    int timerfd = timerfd_create(CLOCK_REALTIME, TFD_NONBLOCK|TFD_CLOEXEC);
    if(timerfd == -1) {
        perror("timerfd_create");
        abort();
    }
    timerevent.events = EPOLLIN;
    timerevent.data.fd = timerfd;
    timerits.it_interval.tv_sec = 0;
    timerits.it_interval.tv_nsec = 250 * 1000 * 1000;
    timerits.it_value.tv_sec = timerits.it_interval.tv_sec;
    timerits.it_value.tv_nsec = timerits.it_interval.tv_nsec;

    client = ixp_nsmount("wmii");
    if(client == NULL) {
        printf("ixp_nsmount: %s\n", ixp_errbuf());
        abort();
    }

    res = epoll_ctl(epollfd, EPOLL_CTL_ADD, timerfd, &timerevent);
    if(res == -1) {
        perror("epoll_ctl");
        abort();
    }
        
    res = timerfd_settime(timerfd, 0, &timerits, NULL);
    if(res == -1) {
        perror("timerfd_settime");
        abort();
    }

    sb_init(&sb, client);
    sb_add(&sb, &sbe_sda);
    sb_add(&sb, &sbe_sdb);
    sb_add(&sb, &sbe_sdc);

    while(1) {
        nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);
        if(nfds == -1) {
		    if(errno != EINTR) {
                perror("epoll_wait");
                abort();
			}
        }

        if(should_quit) {
            break;
        }
        
        for (n = 0; n < nfds; n++) {
            if(events[n].data.fd == timerfd) {
                uint64_t x;
                read(timerfd, &x, sizeof(x));
                sb_update(&sb);
//.........这里部分代码省略.........

int main(int argc, char *argv[])
{
	int listenfd;
	int longindex = 0;
	int c;
	int count  = 1000000;
	pid_t pid = getpid();

	/* Epoll variables */
	struct epoll_event ev;
	int epollfd;

	/* Default settings */
	int addr_family = AF_INET; /* Default address family */
	uint16_t listen_port = 6666;

	/* Support for both IPv4 and IPv6.
	 *  sockaddr_storage: Can contain both sockaddr_in and sockaddr_in6
	 */
	struct sockaddr_storage listen_addr;

	memset(&listen_addr, 0, sizeof(listen_addr));

	/* Parse commands line args */
	while ((c = getopt_long(argc, argv, "c:l:64swv:",
			long_options, &longindex)) != -1) {
		if (c == 0) { /* optional handling "flag" options */
			if (verbose) {
				printf("Flag option %s",
				       long_options[longindex].name);
				if (optarg) printf(" with arg %s", optarg);
				printf("\n");
			}
		}
		if (c == 'c') count       = atoi(optarg);
		if (c == 'l') listen_port = atoi(optarg);
		if (c == '4') addr_family = AF_INET;
		if (c == '6') addr_family = AF_INET6;
		if (c == 'w') write_something = 1;
		if (c == 'v') (optarg) ? verbose = atoi(optarg) : (verbose = 1);
		if (c == '?') return usage(argv);
	}

	if (verbose > 0)
		printf("IP%s TCP listen port %d PID:[%d]\n",
		       (addr_family == AF_INET6) ? "v6":"v4",
		       listen_port, pid);

	/* Socket setup stuff */
	listenfd = Socket(addr_family, SOCK_STREAM, IPPROTO_IP);

	/* Enable use of SO_REUSEPORT for multi-process testing  */
	if (so_reuseport) {
		if ((setsockopt(listenfd, SOL_SOCKET, SO_REUSEPORT,
				&so_reuseport, sizeof(so_reuseport))) < 0) {
			printf("ERROR: No support for SO_REUSEPORT\n");
			perror("- setsockopt(SO_REUSEPORT)");
			exit(EXIT_FAIL_SOCKOPT);
		} else if (verbose) {
			printf(" - Enabled SO_REUSEPORT\n");
		}
	}

	/* Setup listen_addr depending on IPv4 or IPv6 address */
	//setup_sockaddr(addr_family, &listen_addr, "0.0.0.0", listen_port);
	if (addr_family == AF_INET) {
		struct sockaddr_in *addr4 = (struct sockaddr_in *)&listen_addr;
		addr4->sin_family      = addr_family;
		addr4->sin_port        = htons(listen_port);
		addr4->sin_addr.s_addr = htonl(INADDR_ANY);
	} else if (addr_family == AF_INET6) {
		struct sockaddr_in6 *addr6 = (struct sockaddr_in6 *)&listen_addr;
		addr6->sin6_family= addr_family;
		addr6->sin6_port  = htons(listen_port);
	}

	Bind(listenfd, &listen_addr);

	/* Notice "backlog" limited by: /proc/sys/net/core/somaxconn */
	listen(listenfd, 1024);

	/* Epoll */
	if (use_epoll) {
		epollfd = epoll_create1(0);
		if (epollfd == -1) {
			perror("epoll_create");
			exit(EXIT_FAILURE);
		}

		/* Add listen socket */
		ev.events = EPOLLIN;
		ev.data.fd = listenfd;
		if (epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &ev) == -1) {
			perror(" - epoll_ctl: cannot add listen sock");
			exit(EXIT_FAILURE);
		}

		epoll_connections(epollfd, &ev, listenfd, count);

		close(epollfd);
//.........这里部分代码省略.........

// receive each one block from all sender
void* ExpandableBlockStreamExchangeEpoll::receiver(void* arg){
	ExpandableBlockStreamExchangeEpoll* Pthis=(ExpandableBlockStreamExchangeEpoll*)arg;

	struct epoll_event event;
	struct epoll_event *events;

	int status;

	/** create epoll **/
	Pthis->epoll_fd_ = epoll_create1(0);
	if (Pthis->epoll_fd_ == -1)
	{
		Pthis->logging_->elog("epoll create error!\n");
		return 0;
	}

	event.data.fd = Pthis->sock_fd;
	event.events = EPOLLIN | EPOLLET;
	status = epoll_ctl(Pthis->epoll_fd_, EPOLL_CTL_ADD, Pthis->sock_fd, &event);
	if (status == -1)
	{
		Pthis->logging_->elog("epoll ctl error!\n");
		return 0;
	}


	events=(epoll_event*)calloc(Pthis->nlowers,sizeof(epoll_event));
	int fd_cur=0;
	ticks start=curtick();
	std::vector<int> finish_times;//in ms
	while(true){
		usleep(1);
		const int event_count = epoll_wait(Pthis->epoll_fd_, events, Pthis->nlowers, -1);
		for (int i = 0; i < event_count; i++)
		{
			if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP) || (!(events[i].events & EPOLLIN)))
			{
				if (errno == EINTR)
				{
					continue;
				}
				Pthis->logging_->elog("[%ld] epoll error,reason:%s\n", Pthis->state.exchange_id_, strerror(errno));
				FileClose(events[i].data.fd);
				std::cout << "in " << __FILE__ << ":" << __LINE__;
				printf("-----for debug:close fd %d.\n", events[i].data.fd);
				continue;
			}
			else if (Pthis->sock_fd == events[i].data.fd)
			{
				/* We have a notification on the listening socket, which means one or more incoming connections.*/
				while (true)
				{
					sockaddr in_addr;
					socklen_t in_len;
					int infd;
					char hbuf[NI_MAXHOST], sbuf[NI_MAXSERV];

					in_len = sizeof in_addr;
					infd = accept(Pthis->sock_fd, &in_addr, &in_len);
					if (infd == -1)
					{
						if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
						{
							/* all the incoming connections are processed.*/
							break;
						}
						else
						{
							Pthis->logging_->elog("accept error!  ");
							break;
						}
					}
					status=getnameinfo(&in_addr,in_len,hbuf,sizeof(hbuf),sbuf,sizeof(sbuf),NI_NUMERICHOST|NI_NUMERICSERV);
					if(status==0){
						Pthis->logging_->log("[%ld] Accepted connection on descriptor %d (host=%s, port=%s),id=%d\n",Pthis->state.exchange_id_, infd, hbuf, sbuf,Pthis->state.exchange_id_);
						Pthis->lower_ip_array.push_back(hbuf);
						Pthis->lower_sock_fd_to_index[infd]=Pthis->lower_ip_array.size()-1;
						assert(Pthis->lower_ip_array.size()<=Pthis->state.lower_id_list_.size());
					}
					/*Make the incoming socket non-blocking and add it to the list of fds to monitor.*/
					if (!Pthis->SetSocketNonBlocking(infd))
					{
						return 0;
					}
					event.data.fd = infd;
					event.events = EPOLLIN | EPOLLET;
					status = epoll_ctl(Pthis->epoll_fd_, EPOLL_CTL_ADD, infd, &event);
					if (status == -1)
					{
						Pthis->logging_->elog("epoll_ctl");
						return 0;
					}
				}
				continue;
			}
			else
			{
				/* We have data on the fd waiting to be read.*/
				int done = 0;
//.........这里部分代码省略.........

int main (int argc, char *argv[])
{
	struct epoll_event event, *events;
	int sfd, s;
	int efd;

	struct stat st;
	char *fifo = "event.fifo";

	if (lstat (fifo, &st) == 0) {
		if ((st.st_mode & S_IFMT) == S_IFREG) {
			errno = EEXIST;
			err_sys("lstat");
			exit (1);
		}
	}

	unlink (fifo);
	if (mkfifo (fifo, 0600) == -1) {
		err_sys("mkfifo");
		exit (1);
	}

	/* Linux pipes are broken, we need O_RDWR instead of O_RDONLY */
	sfd = open (fifo, O_RDWR | O_NONBLOCK, 0);

	if (sfd == -1) {
		err_sys("open");
		exit (1);
	}

	s = make_socket_non_blocking (sfd);
	if (s == -1)
	{
		err_sys("socket error");
		exit(1);
	}

	efd = epoll_create1 (0);
	if (efd == -1)
	{
		err_sys("epoll_create");
		exit(1);
	}

	event.data.fd = sfd;
	event.events = EPOLLIN | EPOLLET;
	s = epoll_ctl (efd, EPOLL_CTL_ADD, sfd, &event);
	if (s == -1)
	{
		err_sys("epoll_ctl");
		exit(1);
	}
	/* Buffer where events are returned */
	events = calloc (MAX_EVENTS, sizeof event);
	//events = (struct epoll_event *)calloc (MAX_EVENTS, sizeof event);
	/* The event loop */
	while (1)
	{
		int n, i;

		n = epoll_wait (efd, events, MAX_EVENTS, -1);
		for (i = 0; i < n; i++)
		{
			if ((events[i].events & EPOLLERR) ||
					(events[i].events & EPOLLHUP) ||
					(!(events[i].events & EPOLLIN)))
			{
				/* An error has occured on this fd, or the socket is not
				   ready for reading (why were we notified then?) */
				fprintf (stderr, "epoll error\n");
				close (events[i].data.fd);
				continue;
			}
			else
			{
				/* We have data on the fd waiting to be read. Read and
				   display it. We must read whatever data is available
				   completely, as we are running in edge-triggered mode
				   and won't get a notification again for the same
				   data. */
				while (1)
				{
					ssize_t count;
					char buf[2];

					count = read (events[i].data.fd, buf, sizeof buf);
					if (count == -1)
					{
						/* If errno == EAGAIN, that means we have read all
						   data. So go back to the main loop. */
						if (errno != EAGAIN)
						{
							err_sys("read");
						}
						break;
					}
					/* Write the buffer to standard output */
					s = write (1, buf, count);
					if (s == -1)
//.........这里部分代码省略.........

int main(int argc, char *argv[])
{
    struct epoll_event ev;
    struct epoll_event *evs;

    if (argc != 2)
    {
        fprintf(stderr, "Usage: epoll port\n");
        exit(1);
    }

    int socketfd = create_socket_and_bind_port(argv[1]);
    if (socketfd == -1)
    {
        exit(EXIT_FAILURE);
    }

    if (make_fd_nonblock(socketfd) == -1)
    {
        exit(EXIT_FAILURE);
    }

    if (listen_socket(socketfd, MAXLISTEN) == -1)
    {
        exit(EXIT_FAILURE);
    }

    int epollfd;
    if ((epollfd = epoll_create1(0)) == -1)
    {
        perror("epoll_create1");
        exit(EXIT_FAILURE);
    }

    ev.events    = EPOLLIN | EPOLLET;
    ev.data.fd   = socketfd;
    if (epoll_ctl(epollfd, EPOLL_CTL_ADD, socketfd, &ev) == -1)
    {
        perror("epoll_ctl"); 
        exit(EXIT_FAILURE);
    }

    evs= calloc(sizeof (struct epoll_event), MAXEVENTS);
    if (!evs)
    {
        perror("calloc");
        exit(EXIT_FAILURE);
    }
    
    for ( ; ;)
    {
        int nret; 
        nret = epoll_wait(epollfd, evs, MAXEVENTS, -1);
        int i;
        for (i = 0; i < nret; ++i)
        {
            ev = evs[i];
            if ((ev.events & EPOLLERR) || 
                 (ev.events & EPOLLHUP) || 
                (!(ev.events & EPOLLIN)))
            {
                fprintf(stderr, "epoll error: %u\n", ev.events); 
                epoll_ctl(epollfd, EPOLL_CTL_DEL, ev.data.fd, NULL);
                close(ev.data.fd); 
                continue;
            }

            if (ev.data.fd == socketfd)
            {  
                for (; ;)
                {
                    int newfd;
                    struct sockaddr sa;
                    socklen_t sa_len = sizeof(struct sockaddr);
                    newfd = accept(socketfd, &sa, &sa_len);
                    if (newfd == -1)
                    {
                        if ( errno == EAGAIN || errno == EWOULDBLOCK)
                        {
                            break;
                        }
                        perror("accept");
                        break;
                    }

                    char ip[INET6_ADDRSTRLEN];
                    void *addr = get_in_addr(&sa);
                    inet_ntop(sa.sa_family, addr, ip, sizeof(ip));
                    printf("Accept connection %s on descriptor %d\n", ip, newfd);
                    
                    if (make_fd_nonblock(newfd) == -1)
                    {
                        exit(EXIT_FAILURE);
                    }

                    struct epoll_event newev;
                    newev.data.fd = newfd;
                    newev.events = EPOLLIN | EPOLLET;
                    if (epoll_ctl(epollfd, EPOLL_CTL_ADD, newfd, &newev) == -1)
                    {
//.........这里部分代码省略.........

io_service::io_service() {
  int flags = 0;
  efd = epoll_create1(flags);
}

static int server_init(Server *s, unsigned n_sockets) {
        int r;
        unsigned i;

        assert(s);
        assert(n_sockets > 0);

        zero(*s);

        s->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
        if (s->epoll_fd < 0) {
                r = log_error_errno(errno,
                                    "Failed to create epoll object: %m");
                goto fail;
        }

        for (i = 0; i < n_sockets; i++) {
                struct epoll_event ev;
                Fifo *f;
                int fd;

                fd = SD_LISTEN_FDS_START+i;

                r = sd_is_fifo(fd, NULL);
                if (r < 0) {
                        log_error_errno(r, "Failed to determine file descriptor type: %m");
                        goto fail;
                }

                if (!r) {
                        log_error("Wrong file descriptor type.");
                        r = -EINVAL;
                        goto fail;
                }

                f = new0(Fifo, 1);
                if (!f) {
                        r = -ENOMEM;
                        log_error_errno(errno, "Failed to create fifo object: %m");
                        goto fail;
                }

                f->fd = -1;

                zero(ev);
                ev.events = EPOLLIN;
                ev.data.ptr = f;
                if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, fd, &ev) < 0) {
                        r = -errno;
                        fifo_free(f);
                        log_error_errno(errno, "Failed to add fifo fd to epoll object: %m");
                        goto fail;
                }

                f->fd = fd;
                LIST_PREPEND(fifo, s->fifos, f);
                f->server = s;
                s->n_fifos++;
        }

        r = bus_connect_system_systemd(&s->bus);
        if (r < 0) {
                log_error_errno(r, "Failed to get D-Bus connection: %m");
                r = -EIO;
                goto fail;
        }

        return 0;

fail:
        server_done(s);

        return r;
}

/* Handles a number of connections for a thread.
 *
 * data  The thread data.
 */
static void *ThreadHandler(void *data)
{
    int                 ret;
    socklen_t           socketfd = -1;
    int                 efd;
    struct epoll_event  event;
    struct epoll_event  event_conn;
    struct epoll_event* events = NULL;
    ThreadData*         threadData = (ThreadData*)data;
#ifdef WOLFSSL_ASYNC_CRYPT
    WOLF_EVENT*         wolfEvents[MAX_WOLF_EVENTS];
#endif

    /* Initialize wolfSSL and create a context object. */
    if (WolfSSLCtx_Init(version, ourCert, ourKey, verifyCert, cipherList,
                        &threadData->devId, &threadData->ctx) == -1) {
        exit(EXIT_FAILURE);
    }

    /* Allocate space for EPOLL events to be stored. */
    events = (struct epoll_event*)malloc(EPOLL_NUM_EVENTS * sizeof(*events));
    if (events == NULL)
        exit(EXIT_FAILURE);

    /* Create a socket and listen for a client. */
    if (CreateSocketListen(port, numClients, &socketfd) == EXIT_FAILURE)
        exit(EXIT_FAILURE);

    /* Create an EPOLL file descriptor. */
    efd = epoll_create1(0);
    if (efd == -1) {
        fprintf(stderr, "ERROR: failed to create epoll\n");
        exit(EXIT_FAILURE);
    }

    /* Add the event for communications on listening socket. */
    memset(&event, 0, sizeof(event));
    event.events = EPOLLIN;
    event.data.ptr = NULL;
    ret = epoll_ctl(efd, EPOLL_CTL_ADD, socketfd, &event);
    if (ret == -1) {
        fprintf(stderr, "ERROR: failed to add event to epoll\n");
        exit(EXIT_FAILURE);
    }
    threadData->accepting = 1;

    /* Keep handling clients until done. */
    while (!SSLConn_Done(sslConnCtx)) {
        int n;
        int i;

#ifdef WOLFSSL_ASYNC_CRYPT
        do {
            double diff, start = current_time(1);
            ret = wolfSSL_CTX_AsyncPoll(threadData->ctx, wolfEvents,
                                        MAX_WOLF_EVENTS,
                                        WOLF_POLL_FLAG_CHECK_HW, &n);
            diff = current_time(0) - start;
            pthread_mutex_lock(&sslConnMutex);
            sslConnCtx->asyncTime += diff;
            pthread_mutex_unlock(&sslConnMutex);
            for (i = 0; i < n; i++) {
                SSLConn* sslConn = threadData->sslConn;

                while (sslConn != NULL) {
                    if (sslConn->ssl != wolfEvents[i]->context) {
                        sslConn = sslConn->next;
                        continue;
                    }

                    SSLConn_ReadWrite(sslConnCtx, threadData, sslConn);
                    break;
                }
            }
        } while (n > 0);
#endif

        SSLConn_FreeSSLConn(threadData);

#ifdef WOLFSSL_ASYNC_CRYPT
        /* Look for events. */
        n = epoll_wait(efd, events, EPOLL_NUM_EVENTS, 0);
#else
        /* Wait a second for events. */
        n = epoll_wait(efd, events, EPOLL_NUM_EVENTS, 1);
#endif
        /* Process all returned events. */
        for (i = 0; i < n; i++) {
            /* Error event on socket. */
            if (!(events[i].events & EPOLLIN)) {
                if (events[i].data.ptr == NULL) {
                    /* Not a client, therefore the listening connection. */
                    close(socketfd);
                    socketfd = -1;
                }
                else {
//.........这里部分代码省略.........

static void lwan_fd_watch_init(struct lwan *l)
{
    l->epfd = epoll_create1(EPOLL_CLOEXEC);
    if (l->epfd < 0)
        lwan_status_critical_perror("epoll_create1");
}

static apr_status_t impl_pollset_create(apr_pollset_t *pollset,
                                        apr_uint32_t size,
                                        apr_pool_t *p,
                                        apr_uint32_t flags)
{
    apr_status_t rv;
    int fd;

#ifdef HAVE_EPOLL_CREATE1
    fd = epoll_create1(EPOLL_CLOEXEC);
#else
    fd = epoll_create(size);
#endif
    if (fd < 0) {
        pollset->p = NULL;
        return apr_get_netos_error();
    }

#ifndef HAVE_EPOLL_CREATE1
    {
        int fd_flags;

        if ((fd_flags = fcntl(fd, F_GETFD)) == -1) {
            rv = errno;
            close(fd);
            pollset->p = NULL;
            return rv;
        }

        fd_flags |= FD_CLOEXEC;
        if (fcntl(fd, F_SETFD, fd_flags) == -1) {
            rv = errno;
            close(fd);
            pollset->p = NULL;
            return rv;
        }
    }
#endif

    pollset->p = apr_palloc(p, sizeof(apr_pollset_private_t));
#if APR_HAS_THREADS
    if ((flags & APR_POLLSET_THREADSAFE) &&
        !(flags & APR_POLLSET_NOCOPY) &&
        ((rv = apr_thread_mutex_create(&pollset->p->ring_lock,
                                       APR_THREAD_MUTEX_DEFAULT,
                                       p)) != APR_SUCCESS)) {
        close(fd);
        pollset->p = NULL;
        return rv;
    }
#else
    if (flags & APR_POLLSET_THREADSAFE) {
        close(fd);
        pollset->p = NULL;
        return APR_ENOTIMPL;
    }
#endif
    pollset->p->epoll_fd = fd;
    pollset->p->pollset = apr_palloc(p, size * sizeof(struct epoll_event));
    pollset->p->result_set = apr_palloc(p, size * sizeof(apr_pollfd_t));

    if (!(flags & APR_POLLSET_NOCOPY)) {
        APR_RING_INIT(&pollset->p->query_ring, pfd_elem_t, link);
        APR_RING_INIT(&pollset->p->free_ring, pfd_elem_t, link);
        APR_RING_INIT(&pollset->p->dead_ring, pfd_elem_t, link);
    }
    return APR_SUCCESS;
}

	PollEpoll::PollEpoll():activeEv_(kMaxEvents)
	{
		evId_ = epoll_create1(EPOLL_CLOEXEC);
	}

void Engine::init(const char *fname)
{
	DLOG(INFO) << "Loading configuration";
	try
	{
		config.readFile(fname);
	}
	catch (const FileIOException &fioex)
	{
		throw std::runtime_error("I/O error while reading file.");
	}
	catch (const ParseException &pex)
	{
		throw std::runtime_error("Parse configuration file error");
	}

	//Load server configuration here
	int number_of_worker;
	Setting &sconf = config.getRoot()["general"];
	if (!sconf.lookupValue("number_of_worker", number_of_worker))
	{
		number_of_worker = 10;
		DLOG(ERROR) << "Fail to load number_of_worker parameter from configuration file";
	}

	for (int i = 0; i < number_of_worker; i++)
	{
		boost::shared_ptr<Worker> w(new Worker(i));
		workers.push_back(w);
	}

	int task_queue_size;
	if (!sconf.lookupValue("task_queue_size", task_queue_size))
	{
		task_queue_size = 100;
		DLOG(ERROR) << "Use default EventQueue size: 100";
	}
	tasks.init(task_queue_size);

	if (!sconf.lookupValue("max_event", max_event))
	{
		max_event = 512;
		DLOG(ERROR) << "Use default max_event: 512";
	}

	/* register services */
	REGISTER_SERVICE(CoreService);

	/* create epoll */
	epoll_fd = epoll_create1(0);
	if (epoll_fd == -1)
	{
		throw std::runtime_error("Error in epoll_create");
	}

	/* init components */
	for (component_map::iterator it = components.begin(); it != components.end(); it++)
	{
		DLOG(INFO) << "Component:" << it->second->get_id();
		it->second->init();
	}
}

void str_cli(FILE *fp, int sockfd)
{
  char sendline[MAXLINE], recvline[MAXLINE];
  int epollfd = epoll_create1(EPOLL_CLOEXEC);
  struct epoll_event event1;
  bzero(&event1, sizeof(event1));
  event1.data.fd = sockfd;
  event1.events = EPOLLIN | EPOLLET;   // 设为 et 模式
  if (epoll_ctl(epollfd, EPOLL_CTL_ADD, sockfd, &event1))
  {
    perror("epoll_ctl: sockfd");
  }
  setnonblocking(sockfd);

  printf("file fd: %d\n", fileno(fp));
  struct epoll_event event2;
  bzero(&event2, sizeof(event2));
  event2.data.fd = fileno(fp);
  event2.events = EPOLLIN | EPOLLET;
  if (epoll_ctl(epollfd, EPOLL_CTL_ADD, fileno(fp), &event2) < 0)
  {
    perror("epoll_ctl: fp");
  }
  printf("file fd: %d\n", fileno(fp));
  setnonblocking(fileno(fp));
  
  struct epoll_event events[events_size];
  for ( ; ; )
  {
    int eventNums = epoll_wait(epollfd, events, events_size, timeoutMs);
    printf("after epoll\n");
    int i, ret = 0, n = 0;
    for (i = 0; i < eventNums; i ++)
    {
      ret = n = 0;
      // socket 可读, et 模式下这次事件不会重复触发, 必须循环全部读取出来
      if (events[i].data.fd == sockfd)
      {
        printf("sockfd ...\n");
        bzero(recvline, MAXLINE);
        while ((ret = read(sockfd, recvline+n, MAXLINE)) > 0)
        {
          n += ret;
          printf("after sockfd read, ret = %d, n = %d; ...\n", ret, n);
        }
	    if (ret < 0 && errno != EAGAIN && errno != EWOULDBLOCK)
        {
          perror("str_cli: server terminated prematurely");
	    }
        fputs(recvline, stdout);
      }
      // 终端数据可读
      else if (events[i].data.fd == fileno(fp))
      {
        printf("terminal ...\n");
        if (fgets(sendline, MAXLINE, fp) == NULL)
        {
          perror("fgets error");
        }
        n = strlen(sendline);
        while (n > 0)
        {
          ret = write(sockfd, sendline+ret, n);
          printf("after sockfd write, ret = %d, n = %d; ...\n", ret, n);
          if (ret < n)
          {
            if (ret < 0 && errno != EAGAIN && errno != EWOULDBLOCK)
            {
              perror("str_cli: write error");
            }
            break;
          }
          n -= ret;
        }
      }
    }
  }
}

int main(int argc, char *argv[]) {
    printf("%s", "start\n");
    int listen_fd;
    int rcode;
    struct epoll_event *events;
    
    if (argc != 2) {
        fprintf(stderr, "usage: %s [port]\n", argv[0]);
        exit(EXIT_FAILURE);
    }
    
    struct sigaction sa;
    memset(&sa, 0, sizeof(sa));
    sa.sa_handler = SIG_IGN;
    sa.sa_flags = 0;
    if (sigaction(SIGPIPE, &sa, NULL)) {
        printf("ignore SIGPIPE\n");
    }
    
    struct sockaddr_in client_addr;
    socklen_t client_len = 1;
    memset(&client_addr, 0, sizeof(struct sockaddr_in));
    
    /* create and bind the port, and then set the socket to non blocking mode */
    listen_fd = open_listenfd(atoi(argv[1]));
    debug("listen fd = %d", listen_fd);
    rcode = make_socket_non_blocking(listen_fd);
    if (rcode == -1) {
        log_err("error when making socket non blocking");
        abort();
    }
    
    /* create epoll event */
    int efd = epoll_create1(0);
    if (efd == -1) {
        log_err("epoll_create");
        abort();
    }
    struct epoll_event event;

    events = (struct epoll_event *)malloc(sizeof(struct epoll_event) * MAXEVENTS);
    
    http_request_t *request = (http_request_t *)malloc(sizeof(http_request_t));
    http_request_init(request, listen_fd);
    
    event.data.ptr = (void *)request;
    event.events = EPOLLIN | EPOLLET;
    
    /* register the listen event */
    rcode = epoll_ctl(efd, EPOLL_CTL_ADD, listen_fd, &event);
    if (rcode == -1) {
        perror("epoll_ctl");
        abort();
    }
    
    threadpool_t *tp = threadpool_init(NUM_OF_THREADS);
    
    /* event loop */
    while (1) {
        int n = epoll_wait(efd, events, MAXEVENTS, -1);
        
        /* process each incoming IO event */
        int i;
        for (i = 0; i < n; i++) {
            http_request_t *r = (http_request_t *)events[i].data.ptr;
            int fd = r->fd;
            debug("event fd = %d", fd);
            if (fd == listen_fd) {  /* incoming connection event */
                
                while (1) {
                    int client_fd;
                    debug("waiting for accept");
                    client_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &client_len);
                    
                    if (client_fd == -1) {
                        if (errno == EAGAIN || errno == EWOULDBLOCK) {
                            // we have already processed the incoming connection
                            debug("incoming connection processed\n");
                            break;
                        }
                        else {
                            log_err("error occured when accepting connection\n");
                            break;
                        }
                    }
                    
                    rcode = make_socket_non_blocking(client_fd);
                    if (rcode == -1) {
                        if (errno == EAGAIN || errno == EWOULDBLOCK)
                            // we have already processed the incoming connection
                            break;
                        log_err("fail to accept the connection\n");
                        break;
                    }
                    debug("new connection fd %d", client_fd);
                    
                    http_request_t *request = (http_request_t *)malloc(sizeof(http_request_t));
                    http_request_init(request, client_fd);
                    event.data.ptr = (void *)request;
                    event.events = EPOLLIN | EPOLLET;
//.........这里部分代码省略.........

void *runServer(void* value)
{
    int i = 0;

    Settings *settings = (Settings *)value;
    struct sockaddr_in sockAddr;
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_port = htons(settings->port);

    int master = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (0 >= inet_pton(AF_INET, settings->ip, &sockAddr.sin_addr.s_addr))
        fprintf(stderr,"ERROR: wrong IP address\n");

    bind(master, (struct sockaddr *)(&sockAddr), sizeof(sockAddr));
    setNonblock(master);
    listen(master, SOMAXCONN);

    int epoll = epoll_create1(0);

    struct epoll_event epollEvent;
    epollEvent.data.fd = master;
    epollEvent.events = EPOLLIN;
    epoll_ctl(epoll, EPOLL_CTL_ADD, master, &epollEvent);

    while(1) {
        struct epoll_event events[EPOLL_SIZE];

        int numEvents = epoll_wait(epoll, events, EPOLL_SIZE, -1);

        for (i = 0; i < numEvents; ++i) {
            if (events[i].data.fd == master) {
                int slave = accept (master, 0, 0);
                struct epoll_event event;
                event.data.fd = slave;
                event.events = EPOLLIN;

                setNonblock (slave);
                epoll_ctl(epoll, EPOLL_CTL_ADD, slave, &event);
            } else {
                char query[BUF_LEN];
                memset(query, 0, sizeof(query));
                if (0 < recv(events[i].data.fd, query, BUF_LEN, MSG_NOSIGNAL)) {
                    int start = 0;
                    int queryLen = strlen(query);

                    toLog("query src:   ", query);

                    while((query[start] != '/') && (start < queryLen))