3 from eventlet import patcher
4 from eventlet.support import six
5 select = patcher.original('select')
6 time = patcher.original('time')
9 from eventlet.support import clear_sys_exc_info
10 from eventlet.hubs.hub import BaseHub, READ, WRITE, noop
13 if getattr(select, 'kqueue', None) is None:
14 raise ImportError('No kqueue implementation found in select module')
17 FILTERS = {READ: select.KQ_FILTER_READ,
18 WRITE: select.KQ_FILTER_WRITE}
24 def __init__(self, clock=time.time):
25 super(Hub, self).__init__(clock)
29 def _init_kqueue(self):
30 self.kqueue = select.kqueue()
31 self._pid = os.getpid()
33 def _reinit_kqueue(self):
37 events = [e for i in six.itervalues(self._events)
38 for e in six.itervalues(i)]
39 kqueue.control(events, 0, 0)
41 def _control(self, events, max_events, timeout):
43 return self.kqueue.control(events, max_events, timeout)
44 except (OSError, IOError):
46 if os.getpid() != self._pid:
48 return self.kqueue.control(events, max_events, timeout)
51 def add(self, evtype, fileno, cb, tb, mac):
52 listener = super(Hub, self).add(evtype, fileno, cb, tb, mac)
53 events = self._events.setdefault(fileno, {})
54 if evtype not in events:
56 event = select.kevent(fileno, FILTERS.get(evtype), select.KQ_EV_ADD)
57 self._control([event], 0, 0)
58 events[evtype] = event
60 super(Hub, self).remove(listener)
64 def _delete_events(self, events):
66 select.kevent(e.ident, e.filter, select.KQ_EV_DELETE)
69 self._control(del_events, 0, 0)
71 def remove(self, listener):
72 super(Hub, self).remove(listener)
73 evtype = listener.evtype
74 fileno = listener.fileno
75 if not self.listeners[evtype].get(fileno):
76 event = self._events[fileno].pop(evtype)
78 self._delete_events([event])
82 def remove_descriptor(self, fileno):
83 super(Hub, self).remove_descriptor(fileno)
85 events = self._events.pop(fileno).values()
86 self._delete_events(events)
92 def wait(self, seconds=None):
93 readers = self.listeners[READ]
94 writers = self.listeners[WRITE]
96 if not readers and not writers:
100 result = self._control([], self.MAX_EVENTS, seconds)
101 SYSTEM_EXCEPTIONS = self.SYSTEM_EXCEPTIONS
104 evfilt = event.filter
106 if evfilt == FILTERS[READ]:
107 readers.get(fileno, noop).cb(fileno)
108 if evfilt == FILTERS[WRITE]:
109 writers.get(fileno, noop).cb(fileno)
110 except SYSTEM_EXCEPTIONS:
113 self.squelch_exception(fileno, sys.exc_info())