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