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