python/ovs/poller.py

   1 # Copyright (c) 2010 Nicira, Inc.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at:
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import errno
  16 import ovs.timeval
  17 import ovs.vlog
  18 import select
  19 import socket
  20
  21 try:
  22     import eventlet.patcher
  23
  24     def _using_eventlet_green_select():
  25         return eventlet.patcher.is_monkey_patched(select)
  26 except:
  27     def _using_eventlet_green_select():
  28         return False
  29
  30 vlog = ovs.vlog.Vlog("poller")
  31
  32
  33 # eventlet/gevent doesn't support select.poll. If select.poll is used,
  34 # python interpreter is blocked as a whole instead of switching from the
  35 # current thread that is about to block to other runnable thread.
  36 # So emulate select.poll by select.select because using python means that
  37 # performance isn't so important.
  38 class _SelectSelect(object):
  39     """ select.poll emulation by using select.select.
  40     Only register and poll are needed at the moment.
  41     """
  42     def __init__(self):
  43         self.rlist = []
  44         self.wlist = []
  45         self.xlist = []
  46
  47     def register(self, fd, events):
  48         if isinstance(fd, socket.socket):
  49             fd = fd.fileno()
  50         assert isinstance(fd, int)
  51         if events & select.POLLIN:
  52             self.rlist.append(fd)
  53             events &= ~select.POLLIN
  54         if events & select.POLLOUT:
  55             self.wlist.append(fd)
  56             events &= ~select.POLLOUT
  57         if events:
  58             self.xlist.append(fd)
  59
  60     def poll(self, timeout):
  61         if timeout == -1:
  62             # epoll uses -1 for infinite timeout, select uses None.
  63             timeout = None
  64         else:
  65             timeout = float(timeout) / 1000
  66         # XXX workaround a bug in eventlet
  67         # see https://github.com/eventlet/eventlet/pull/25
  68         if timeout == 0 and _using_eventlet_green_select():
  69             timeout = 0.1
  70
  71         rlist, wlist, xlist = select.select(self.rlist, self.wlist, self.xlist,
  72                                             timeout)
  73         # collections.defaultdict is introduced by python 2.5 and
  74         # XenServer uses python 2.4. We don't use it for XenServer.
  75         # events_dict = collections.defaultdict(int)
  76         # events_dict[fd] |= event
  77         events_dict = {}
  78         for fd in rlist:
  79             events_dict[fd] = events_dict.get(fd, 0) | select.POLLIN
  80         for fd in wlist:
  81             events_dict[fd] = events_dict.get(fd, 0) | select.POLLOUT
  82         for fd in xlist:
  83             events_dict[fd] = events_dict.get(fd, 0) | (select.POLLERR |
  84                                                         select.POLLHUP |
  85                                                         select.POLLNVAL)
  86         return events_dict.items()
  87
  88
  89 SelectPoll = _SelectSelect
  90 # If eventlet/gevent isn't used, we can use select.poll by replacing
  91 # _SelectPoll with select.poll class
  92 # _SelectPoll = select.poll
  93
  94
  95 class Poller(object):
  96     """High-level wrapper around the "poll" system call.
  97
  98     Intended usage is for the program's main loop to go about its business
  99     servicing whatever events it needs to.  Then, when it runs out of immediate
 100     tasks, it calls each subordinate module or object's "wait" function, which
 101     in turn calls one (or more) of the functions Poller.fd_wait(),
 102     Poller.immediate_wake(), and Poller.timer_wait() to register to be awakened
 103     when the appropriate event occurs.  Then the main loop calls
 104     Poller.block(), which blocks until one of the registered events happens."""
 105
 106     def __init__(self):
 107         self.__reset()
 108
 109     def fd_wait(self, fd, events):
 110         """Registers 'fd' as waiting for the specified 'events' (which should
 111         be select.POLLIN or select.POLLOUT or their bitwise-OR).  The following
 112         call to self.block() will wake up when 'fd' becomes ready for one or
 113         more of the requested events.
 114
 115         The event registration is one-shot: only the following call to
 116         self.block() is affected.  The event will need to be re-registered
 117         after self.block() is called if it is to persist.
 118
 119         'fd' may be an integer file descriptor or an object with a fileno()
 120         method that returns an integer file descriptor."""
 121         self.poll.register(fd, events)
 122
 123     def __timer_wait(self, msec):
 124         if self.timeout < 0 or msec < self.timeout:
 125             self.timeout = msec
 126
 127     def timer_wait(self, msec):
 128         """Causes the following call to self.block() to block for no more than
 129         'msec' milliseconds.  If 'msec' is nonpositive, the following call to
 130         self.block() will not block at all.
 131
 132         The timer registration is one-shot: only the following call to
 133         self.block() is affected.  The timer will need to be re-registered
 134         after self.block() is called if it is to persist."""
 135         if msec <= 0:
 136             self.immediate_wake()
 137         else:
 138             self.__timer_wait(msec)
 139
 140     def timer_wait_until(self, msec):
 141         """Causes the following call to self.block() to wake up when the
 142         current time, as returned by ovs.timeval.msec(), reaches 'msec' or
 143         later.  If 'msec' is earlier than the current time, the following call
 144         to self.block() will not block at all.
 145
 146         The timer registration is one-shot: only the following call to
 147         self.block() is affected.  The timer will need to be re-registered
 148         after self.block() is called if it is to persist."""
 149         now = ovs.timeval.msec()
 150         if msec <= now:
 151             self.immediate_wake()
 152         else:
 153             self.__timer_wait(msec - now)
 154
 155     def immediate_wake(self):
 156         """Causes the following call to self.block() to wake up immediately,
 157         without blocking."""
 158         self.timeout = 0
 159
 160     def block(self):
 161         """Blocks until one or more of the events registered with
 162         self.fd_wait() occurs, or until the minimum duration registered with
 163         self.timer_wait() elapses, or not at all if self.immediate_wake() has
 164         been called."""
 165         try:
 166             try:
 167                 events = self.poll.poll(self.timeout)
 168                 self.__log_wakeup(events)
 169             except select.error, e:
 170                 # XXX rate-limit
 171                 error, msg = e
 172                 if error != errno.EINTR:
 173                     vlog.err("poll: %s" % e[1])
 174         finally:
 175             self.__reset()
 176
 177     def __log_wakeup(self, events):
 178         if not events:
 179             vlog.dbg("%d-ms timeout" % self.timeout)
 180         else:
 181             for fd, revents in events:
 182                 if revents != 0:
 183                     s = ""
 184                     if revents & select.POLLIN:
 185                         s += "[POLLIN]"
 186                     if revents & select.POLLOUT:
 187                         s += "[POLLOUT]"
 188                     if revents & select.POLLERR:
 189                         s += "[POLLERR]"
 190                     if revents & select.POLLHUP:
 191                         s += "[POLLHUP]"
 192                     if revents & select.POLLNVAL:
 193                         s += "[POLLNVAL]"
 194                     vlog.dbg("%s on fd %d" % (s, fd))
 195
 196     def __reset(self):
 197         self.poll = SelectPoll()
 198         self.timeout = -1