def test_add(self):
        import select

        client, server = self.socket_pair()

        ep = select.epoll(2)
        ep.register(server, select.EPOLLIN | select.EPOLLOUT)
        ep.register(client, select.EPOLLIN | select.EPOLLOUT)
        ep.close()

        # adding by object w/ fileno works, too.
        ep = select.epoll(2)
        ep.register(server.fileno(), select.EPOLLIN | select.EPOLLOUT)
        ep.register(client.fileno(), select.EPOLLIN | select.EPOLLOUT)
        ep.close()

        ep = select.epoll(2)
        # TypeError: argument must be an int, or have a fileno() method.
        raises(TypeError, ep.register, object(), select.EPOLLIN | select.EPOLLOUT)
        raises(TypeError, ep.register, None, select.EPOLLIN | select.EPOLLOUT)
        # ValueError: file descriptor cannot be a negative integer (-1)
        raises(ValueError, ep.register, -1, select.EPOLLIN | select.EPOLLOUT)
        # IOError: [Errno 9] Bad file descriptor
        raises(IOError, ep.register, 10000, select.EPOLLIN | select.EPOLLOUT)
        # registering twice also raises an exception
        ep.register(server, select.EPOLLIN | select.EPOLLOUT)
        raises(IOError, ep.register, server, select.EPOLLIN | select.EPOLLOUT)
        ep.close()

    def test_add(self):
        server, client = self._connected_pair()

        ep = select.epoll(2)
        try:
            ep.register(server.fileno(), select.EPOLLIN | select.EPOLLOUT)
            ep.register(client.fileno(), select.EPOLLIN | select.EPOLLOUT)
        finally:
            ep.close()

        # adding by object w/ fileno works, too.
        ep = select.epoll(2)
        try:
            ep.register(server, select.EPOLLIN | select.EPOLLOUT)
            ep.register(client, select.EPOLLIN | select.EPOLLOUT)
        finally:
            ep.close()

        ep = select.epoll(2)
        try:
            # TypeError: argument must be an int, or have a fileno() method.
            self.assertRaises(TypeError, ep.register, object(), select.EPOLLIN | select.EPOLLOUT)
            self.assertRaises(TypeError, ep.register, None, select.EPOLLIN | select.EPOLLOUT)
            # ValueError: file descriptor cannot be a negative integer (-1)
            self.assertRaises(ValueError, ep.register, -1, select.EPOLLIN | select.EPOLLOUT)
            # IOError: [Errno 9] Bad file descriptor
            self.assertRaises(IOError, ep.register, 10000, select.EPOLLIN | select.EPOLLOUT)
            # registering twice also raises an exception
            ep.register(server, select.EPOLLIN | select.EPOLLOUT)
            self.assertRaises(IOError, ep.register, server, select.EPOLLIN | select.EPOLLOUT)
        finally:
            ep.close()

def main():
    args = aux.parse_args('Echo server with epoll-processing model.')
    server_socket = aux.listening_socket(args.port)
    server_fileno = server_socket.fileno()

    fileno_to_data = {}
    fileno_to_socket = {}

    read_epoll = select.epoll()
    read_epoll.register(server_fileno, select.EPOLLIN)

    while True:
        for fileno, eventmask in read_epoll.poll(0.05):
            if fileno == server_fileno:
                client_socket, _ = server_socket.accept()
                client_fileno = client_socket.fileno()

                fileno_to_data[client_fileno] = ''
                fileno_to_socket[client_fileno] = client_socket

                read_epoll.register(client_fileno, select.EPOLLIN)
            else:
                client_socket = fileno_to_socket[fileno]

                data = client_socket.recv(1024)
                if not data:
                    read_epoll.unregister(fileno)
                    del fileno_to_data[fileno]
                    del fileno_to_socket[fileno]
                    client_socket.close()
                else:
                    fileno_to_data[fileno] += data

        check_writability = [f for f, d in fileno_to_data.iteritems() if d]
        if not check_writability:
            continue

        write_epoll = select.epoll()
        for fileno in check_writability:
            write_epoll.register(fileno, select.EPOLLOUT)

        for fileno, eventmask in write_epoll.poll(0.05):
            if eventmask & (select.EPOLLERR | select.EPOLLHUP):
                read_epoll.unregister(fileno)
                fileno_to_socket[fileno].close()
                del fileno_to_data[fileno]
                del fileno_to_socket[fileno]
                continue

            client_socket = fileno_to_socket[fileno]
            data = fileno_to_data[fileno]

            n = client_socket.send(data)
            if n > 0:
                fileno_to_data[fileno] = data[n:]

        write_epoll.close()

	def __init__(self, pipe):
		if isinstance(pipe, int):
			fd = self._fd = pipe
			pipe = os.fromfd(pipe)
		else: fd = self._fd = pipe.fileno()
		self._poll_in, self._poll_out = epoll(), epoll()
		self._poll_in.register(fd, EPOLLIN | EPOLLERR | EPOLLHUP)
		self._poll_out.register(fd, EPOLLOUT | EPOLLERR | EPOLLHUP)
		self.close = pipe.close
		self.reads = pipe.read
		self.writes = pipe.write

 def test_create(self):
     try:
         ep = select.epoll(16)
     except OSError as e:
         raise AssertionError(str(e))
     self.assertTrue(ep.fileno() > 0, ep.fileno())
     self.assertTrue(not ep.closed)
     ep.close()
     self.assertTrue(ep.closed)
     self.assertRaises(ValueError, ep.fileno)
     if hasattr(select, "EPOLL_CLOEXEC"):
         select.epoll(select.EPOLL_CLOEXEC).close()
         self.assertRaises(OSError, select.epoll, flags=12356)

 def __init__(self, pipe, leash=None):
     fd = self._fd = pipe.fileno()
     if leash:
         self.__leash = leash  # fd source object, leashed here to stop gc
     self._poll_in, self._poll_out = epoll(), epoll()
     self._poll_in.register(fd, EPOLLIN)
     self._poll_out.register(fd, EPOLLOUT)
     self.close = pipe.close
     try:  # file
         self.reads = pipe.read
         self.writes = pipe.write
     except AttributeError:  # socket
         self.reads = pipe.recv
         self.writes = pipe.send

	def __collect_piece(self, ips, headers, origin_urls):
		
		# create sockets
		fd_2_sock	= {}
		fd_2_req	= {}
		fd_2_data	= {}
	
		for i in xrange(len(ips)):
			sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
			sock.setblocking(0)
			fd_2_sock[sock.fileno()] = sock
			fd_2_req[sock.fileno()]  = Request(ips[i], headers[i], origin_urls[i])
		print "Create %d Sockets Finished!" % len(fd_2_sock.items())
	
		# create listener
		epoll = select.epoll()
	
		# non-blocking connect
		for (fd, sock) in fd_2_sock.items():
			try:
				sock.connect( (fd_2_req[fd].ip, 80) )
			except Exception, e:
				if e.args[0] == 115:
					epoll.register(fd, select.EPOLLOUT)
				else:
					print "[Connection Error] Url %s : %s" % (fd_2_req[fd].origin_url, str(e))

 def __init__(self, addr, port, logic):
     dbgPrint("\n__init__: start!")
     #初始化状态字典
     self.conn_state = {}
     #初始化socket对象，tcp/ipv4 类型
     self.listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
     self.listen_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     self.listen_sock.bind((addr, port))
     self.listen_sock.listen(10)
     #使用setFd设置socket状态,4个头,读写缓冲，将当前scoket状态类放入字典中
     #并打印当前scoket状态
     self.setFd(self.listen_sock)
     self.epoll_sock = select.epoll()
     #注册 listen_sock 至epollin interface
     self.epoll_sock.register(self.listen_sock.fileno(), select.EPOLLIN)
     self.logic = logic
     #初始化状态机字典
     self.sm = {
         "accept" : self.accept2read,
         "read"   : self.read2process,
         "write"  : self.write2read,
         "process": self.process,
         "closing": self.close,
     }
     dbgPrint("\n__init__: end, register no: %s" %
             self.listen_sock.fileno())

 def __init__(self, *, check_args=True):
     super().__init__(check_args=check_args)
     self._running = True
     self._stopping = False
     self._readers = {}
     self._writers = {}
     self._poll = select.epoll()

def wait_for_pin_state_changes(readers_by_name, devices_by_name):
    fds_to_pins = {}

    # Fetch a list of pins to watch, each of which maps to
    # a card reader
    pin_objects_to_watch = build_hardware_pin_map(readers_by_name)


    # Configure the select handler, setting it to watch the
    # filedescriptors that underly the hardware pins.
    # To do this, we grab the filedescriptors that underly the
    # pins, and build a map between filedescriptors and objects
    epoll_handler = select.epoll()
    for pin_num in pin_objects_to_watch:
        handler_object = pin_objects_to_watch[pin_num]['handler_object']
        gpio_pin = pin_objects_to_watch[pin_num]['gpio_pin']

        epoll_handler.register(gpio_pin, select.EPOLLET)
        fds_to_pins[gpio_pin.fileno()] = pin_num

    # Loop forever, waiting for pin state changes, and triggering the
    # pins based on their values
    while True:
        logging.debug("Waiting for event on reader pins")
        events = epoll_handler.poll()
        for filedescriptor, event in events:
            pin_no = fds_to_pins[filedescriptor]
            logging.debug("Got event on pin number %s" % pin_no)
            pin_value = pin_objects_to_watch[pin_no]['gpio_pin'].value
            pin_objects_to_watch[pin_no]['handler_object'].trigger_pin_state_change(
                        pin_value, devices_by_name)

    def run(self):
        self.poller = select.epoll()
        self.pollmask = select.EPOLLIN | select.EPOLLHUP | select.EPOLLERR
        self.poller.register(self.server, self.pollmask)
        while True:
            try:
                fds = self.poller.poll(timeout=self.timeout)
            except:
                return

            for (fd, event) in fds:
                if event & (select.POLLHUP | select.POLLERR):
                    self.handle_error(fd)
                    continue
                if fd == self.server.fileno():
                    self.handle_server()
                else:
                    self.handle_client(fd)

            to_delete = []
            for fileNo, client in self.clients.iteritems():
                inactive_time = seconds() - client[1]
                if inactive_time > self.timeout:
                    to_delete.append(fileNo)

            for fileNo in to_delete:
                self.close_client(fileNo)

    def __init__(self, RequestHandlerClass):
        if not hasattr(self, "ServerName"):
            self.ServerName = "Eloipool"

        self.RequestHandlerClass = RequestHandlerClass

        self.running = False
        self.keepgoing = True
        self.rejecting = False

        self._epoll = select.epoll()
        self._fd = {}
        self.connections = {}

        self._sch = ScheduleDict()
        self._schEH = {}
        if self.schMT:
            self._schLock = threading.Lock()
        else:
            self._schLock = WithNoop

        self.TrustedForwarders = ()

        if self.waker:
            (r, w) = os.pipe()
            o = _Waker(self, r)
            self.register_socket(r, o)
            self.waker = w

 def __init__(self):
     self.epoll = select.epoll()
     # TODO should we set any other masks?
     # http://docs.python.org/library/select.html#epoll-objects
     self.READ_MASK = select.EPOLLIN | select.EPOLLPRI
     self.WRITE_MASK = select.EPOLLOUT
     self.ERR_MASK = select.EPOLLERR | select.EPOLLHUP

def epoll_poller(timeout=0.0, map=None):
    """A poller which uses epoll(), supported on Linux 2.5.44 and newer."""
    if map is None:
        map = socket_map
    pollster = select.epoll()
    if map:
        for fd, obj in map.items():
            flags = 0
            if obj.readable():
                flags |= select.POLLIN | select.POLLPRI
            if obj.writable():
                flags |= select.POLLOUT
            if flags:
                # Only check for exceptions if object was either readable
                # or writable.
                flags |= select.POLLERR | select.POLLHUP | select.POLLNVAL
                pollster.register(fd, flags)
        try:
            r = pollster.poll(timeout)
        except select.error, err:
            if err.args[0] != EINTR:
                raise
            r = []
        for fd, flags in r:
            obj = map.get(fd)
            if obj is None:
                continue
            readwrite(obj, flags) 

    def __init__(self, popenToWrap):
        self._streamLock = threading.Lock()
        self._proc = popenToWrap

        self._stdout = StringIO()
        self._stderr = StringIO()
        self._stdin = StringIO()

        fdout = self._proc.stdout.fileno()
        fderr = self._proc.stderr.fileno()
        self._fdin = self._proc.stdin.fileno()

        self._closedfds = []

        self._poller = select.epoll()
        self._poller.register(fdout, select.EPOLLIN | select.EPOLLPRI)
        self._poller.register(fderr, select.EPOLLIN | select.EPOLLPRI)
        self._poller.register(self._fdin, 0)
        self._fdMap = {fdout: self._stdout,
                       fderr: self._stderr,
                       self._fdin: self._stdin}

        self.stdout = io.BufferedReader(self._streamWrapper(self,
                                        self._stdout, fdout), BUFFSIZE)

        self.stderr = io.BufferedReader(self._streamWrapper(self,
                                        self._stderr, fderr), BUFFSIZE)

        self.stdin = io.BufferedWriter(self._streamWrapper(self,
                                       self._stdin, self._fdin), BUFFSIZE)

        self._returncode = None

        self.blocking = False

def run_main(listen_fd):
    try:
        epoll_fd = select.epoll()
        epoll_fd.register(listen_fd.fileno(), select.EPOLLIN | select.EPOLLERR | select.EPOLLHUP)
        print "listen_fd:%s" % listen_fd.fileno()
    except select.error, msg:
        logger.error(msg)

 def epoll_run(self):
     """async version"""
     try:
         epoll_fd = select.epoll()
         epoll_fd.register(self.server_socket.fileno(), select.EPOLLIN)
     except select.error, msg:
         exit_error(msg)

    def low_memory_watchdog(self):
        self.counts["low_mem_events"] = 0
        self.__watchdog__run = True
        efd = eventfd(0, 0)
        mp = open(self.MEM_PRES_PATH)
        with open(self.EVENT_CONTR_PATH, "w") as cgc:
            cgc.write("%d %d %s" % (efd, mp.fileno(), self.PRESSURE_LEVEL))

        p = epoll()
        p.register(efd, EPOLLIN)

        self.host.log_info('"Low memory" watchdog started(pressure level=%r).' % self.PRESSURE_LEVEL)
        err = "shutdown event"
        while self.__watchdog__run:
            try:
                events = p.poll(2)
                for fd, event in events:
                    if event & EPOLLIN:
                        # In an eventfd, there are always 8 bytes
                        ret = os.read(efd, 64 / 8)
                        num = struct.unpack("Q", ret)
                        break
                else:
                    continue
            except (ValueError, OSError, IOError) as err:
                break
            self.host.log_debug('"Low memory" notification received: %d' % num)
            for callback in self.low_memory_callbacks:
                callback()
            self.counts["low_mem_events"] += 1

        p.close()
        os.close(efd)
        self.host.log_info('"Low memory" watchdog stopped(msg="%s").' % err)

	def __init__(self, paths):
		self.reactor = epoll()
		self.devices = map(Device, paths)
		self.fds = map(op.attrgetter('fd'), self.devices)
		for fd in self.fds: self.reactor.register(fd, EPOLLIN)
		self.packet_size = Event.get_format_size()
		log.debug('DevGroup init: %r, %s'%(self, self.devices))

def server(ip, port):
    """ 利用IO复用在监听并处理客户连接"""
    # 创建监听套接字
    listen_socket = socket.socket()
    serve_address = (ip, port)
    # 设置地址重用和端口重用
    listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
    listen_socket.bind(serve_address)
    listen_socket.listen(1024)

    # 创建epoller对象
    epoller = epoll()

    # 将listen_socket加入到epoller的监听中
    epoller.register(listen_socket, select.EPOLLIN)

    # SIGINT信号的处理函数
    def sigint_handler(sig_num, addtion):
        listen_socket.close()
        epoller.close()
        print('程序被强制退出...')
        sys.exit()
    
    # 注册对退出信号SIGINT的处理
    signal.signal(signal.SIGINT, sigint_handler)

    print('进程', os.getpid(), '已启动')

    # 开始epoll轮询
    epoll_loop(epoller, listen_socket)