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