python-eventlet/eventlet/green/select.py

   1 __select = __import__('select')
   2 error = __select.error
   3 from eventlet.greenthread import getcurrent
   4 from eventlet.hubs import get_hub
   5 from eventlet.support import six
   6
   7
   8 __patched__ = ['select']
   9 # FIXME: must also delete `poll`, but it breaks subprocess `communicate()`
  10 # https://github.com/eventlet/eventlet/issues/290
  11 __deleted__ = ['devpoll', 'epoll', 'kqueue', 'kevent']
  12
  13
  14 def get_fileno(obj):
  15     # The purpose of this function is to exactly replicate
  16     # the behavior of the select module when confronted with
  17     # abnormal filenos; the details are extensively tested in
  18     # the stdlib test/test_select.py.
  19     try:
  20         f = obj.fileno
  21     except AttributeError:
  22         if not isinstance(obj, six.integer_types):
  23             raise TypeError("Expected int or long, got %s" % type(obj))
  24         return obj
  25     else:
  26         rv = f()
  27         if not isinstance(rv, six.integer_types):
  28             raise TypeError("Expected int or long, got %s" % type(rv))
  29         return rv
  30
  31
  32 def select(read_list, write_list, error_list, timeout=None):
  33     # error checking like this is required by the stdlib unit tests
  34     if timeout is not None:
  35         try:
  36             timeout = float(timeout)
  37         except ValueError:
  38             raise TypeError("Expected number for timeout")
  39     hub = get_hub()
  40     timers = []
  41     current = getcurrent()
  42     assert hub.greenlet is not current, 'do not call blocking functions from the mainloop'
  43     ds = {}
  44     for r in read_list:
  45         ds[get_fileno(r)] = {'read': r}
  46     for w in write_list:
  47         ds.setdefault(get_fileno(w), {})['write'] = w
  48     for e in error_list:
  49         ds.setdefault(get_fileno(e), {})['error'] = e
  50
  51     listeners = []
  52
  53     def on_read(d):
  54         original = ds[get_fileno(d)]['read']
  55         current.switch(([original], [], []))
  56
  57     def on_write(d):
  58         original = ds[get_fileno(d)]['write']
  59         current.switch(([], [original], []))
  60
  61     def on_error(d, _err=None):
  62         original = ds[get_fileno(d)]['error']
  63         current.switch(([], [], [original]))
  64
  65     def on_timeout2():
  66         current.switch(([], [], []))
  67
  68     def on_timeout():
  69         # ensure that BaseHub.run() has a chance to call self.wait()
  70         # at least once before timed out.  otherwise the following code
  71         # can time out erroneously.
  72         #
  73         # s1, s2 = socket.socketpair()
  74         # print(select.select([], [s1], [], 0))
  75         timers.append(hub.schedule_call_global(0, on_timeout2))
  76
  77     if timeout is not None:
  78         timers.append(hub.schedule_call_global(timeout, on_timeout))
  79     try:
  80         for k, v in six.iteritems(ds):
  81             if v.get('read'):
  82                 listeners.append(hub.add(hub.READ, k, on_read, on_error, lambda x: None))
  83             if v.get('write'):
  84                 listeners.append(hub.add(hub.WRITE, k, on_write, on_error, lambda x: None))
  85         try:
  86             return hub.switch()
  87         finally:
  88             for l in listeners:
  89                 hub.remove(l)
  90     finally:
  91         for t in timers:
  92             t.cancel()