def testPair(self):
kq = select.kqueue()
a, b = socket.socketpair()
a.send(b'foo')
event1 = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
event2 = select.kevent(b, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
r = kq.control([event1, event2], 1, 1)
self.assertTrue(r)
self.assertFalse(r[0].flags & select.KQ_EV_ERROR)
self.assertEqual(b.recv(r[0].data), b'foo')
a.close()
b.close()
kq.close()
python类KQ_EV_ERROR的实例源码
def testPair(self):
kq = select.kqueue()
a, b = socket.socketpair()
a.send(b'foo')
event1 = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
event2 = select.kevent(b, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
r = kq.control([event1, event2], 1, 1)
self.assertTrue(r)
self.assertFalse(r[0].flags & select.KQ_EV_ERROR)
self.assertEqual(b.recv(r[0].data), b'foo')
a.close()
b.close()
kq.close()
def __add_ev_read(self, fileno):
"""
Note:if the event exists,it will not do anything
"""
if fileno not in self.__rlist and self.__async_mode == "select":
self.__rlist.append(fileno)
if self.__async_mode == "epoll":
if fileno not in self.__epoll_register_info:
self.__epoll_register_info[fileno] = None
eventmask = self.__epoll_register_info[fileno]
event = select.EPOLLIN
if eventmask == None:
eventmask = event
self.__epoll_object.register(fileno, eventmask)
self.__epoll_register_info[fileno] = eventmask
return
is_register_read = (eventmask & select.EPOLLIN) == select.EPOLLIN
if is_register_read == False:
eventmask = event | eventmask
self.__epoll_object.modify(fileno, eventmask)
if self.__async_mode == "kqueue":
filter_ = select.KQ_FILTER_READ
flags = select.KQ_EV_ADD | select.KQ_EV_ERROR | select.KQ_EV_ENABLE
if fileno not in self.__kqueue_event_map:
kevent = select.kevent(fileno, filter_, flags)
kevent.udata = 0
else:
kevent = self.__kqueue_event_map[fileno]
read_exists = (kevent.udata & EV_TYPE_READ) == EV_TYPE_READ
if read_exists == False:
kevent.filter = filter_
kevent.udata = (kevent.udata | EV_TYPE_READ)
kevent.flags = flags
if fileno not in self.__kqueue_change_event_map:
self.__kqueue_change_event_map[fileno] = []
self.__kqueue_change_event_map[fileno].append(kevent)
''''''
return
def __add_ev_write(self, fileno):
if fileno not in self.__wlist and self.__async_mode == "select":
self.__wlist.append(fileno)
if self.__async_mode == "epoll":
if fileno not in self.__epoll_register_info:
self.__epoll_register_info[fileno] = None
eventmask = self.__epoll_register_info[fileno]
event = select.EPOLLOUT
if eventmask == None:
eventmask = event
self.__epoll_object.register(fileno, eventmask)
self.__epoll_register_info[fileno] = eventmask
return
is_register_write = (eventmask & select.EPOLLOUT) == select.EPOLLOUT
if is_register_write == False:
eventmask = event | eventmask
self.__epoll_object.modify(fileno, eventmask)
if self.__async_mode == "kqueue":
filter_ = select.KQ_FILTER_WRITE
flags = select.KQ_EV_ADD | select.KQ_EV_ERROR | select.KQ_EV_ENABLE
if fileno not in self.__kqueue_event_map:
kevent = select.kevent(fileno, filter_, flags)
kevent.udata = 0
else:
kevent = self.__kqueue_event_map[fileno]
write_exists = (kevent.udata & EV_TYPE_WRITE) == EV_TYPE_WRITE
if write_exists == False:
kevent.filter = filter_
kevent.flags = flags
kevent.udata = (kevent.udata | EV_TYPE_WRITE)
if fileno not in self.__kqueue_change_event_map:
self.__kqueue_change_event_map[fileno] = []
self.__kqueue_change_event_map[fileno].append(kevent)
''''''
return
def poll(self,
timeout,
_len=len,
_READ=select.KQ_FILTER_READ,
_WRITE=select.KQ_FILTER_WRITE,
_EOF=select.KQ_EV_EOF,
_ERROR=select.KQ_EV_ERROR):
try:
kevents = self._kqueue.control(None, _len(self.socket_map),
timeout)
except OSError as err:
if err.errno == errno.EINTR:
return
raise
for kevent in kevents:
inst = self.socket_map.get(kevent.ident)
if inst is None:
continue
if kevent.filter == _READ:
if inst.readable():
_read(inst)
if kevent.filter == _WRITE:
if kevent.flags & _EOF:
# If an asynchronous connection is refused,
# kqueue returns a write event with the EOF
# flag set.
# Note that for read events, EOF may be returned
# before all data has been consumed from the
# socket buffer, so we only check for EOF on
# write events.
inst.handle_close()
else:
if inst.writable():
_write(inst)
if kevent.flags & _ERROR:
inst.handle_close()
# ===================================================================
# --- choose the better poller for this platform
# ===================================================================
