python/ovs/poller.py

   1 # Copyright (c) 2010, 2015 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 os
  17
  18 import select
  19 import socket
  20 import sys
  21
  22 import ovs.timeval
  23 import ovs.vlog
  24
  25 if sys.platform == "win32":
  26     import ovs.winutils as winutils
  27
  28 try:
  29     from OpenSSL import SSL
  30 except ImportError:
  31     SSL = None
  32
  33 try:
  34     from eventlet import patcher as eventlet_patcher
  35
  36     def _using_eventlet_green_select():
  37         return eventlet_patcher.is_monkey_patched(select)
  38 except:
  39     eventlet_patcher = None
  40
  41     def _using_eventlet_green_select():
  42         return False
  43
  44 try:
  45     from gevent import monkey as gevent_monkey
  46 except:
  47     gevent_monkey = None
  48
  49
  50 vlog = ovs.vlog.Vlog("poller")
  51
  52 POLLIN = 0x001
  53 POLLOUT = 0x004
  54 POLLERR = 0x008
  55 POLLHUP = 0x010
  56 POLLNVAL = 0x020
  57
  58
  59 # eventlet/gevent doesn't support select.poll. If select.poll is used,
  60 # python interpreter is blocked as a whole instead of switching from the
  61 # current thread that is about to block to other runnable thread.
  62 # So emulate select.poll by select.select because using python means that
  63 # performance isn't so important.
  64 class _SelectSelect(object):
  65     """ select.poll emulation by using select.select.
  66     Only register and poll are needed at the moment.
  67     """
  68     def __init__(self):
  69         self.rlist = []
  70         self.wlist = []
  71         self.xlist = []
  72
  73     def register(self, fd, events):
  74         if isinstance(fd, socket.socket):
  75             fd = fd.fileno()
  76         if SSL and isinstance(fd, SSL.Connection):
  77             fd = fd.fileno()
  78
  79         if sys.platform != 'win32':
  80             # Skip this on Windows, it also register events
  81             assert isinstance(fd, int)
  82         if events & POLLIN:
  83             self.rlist.append(fd)
  84             events &= ~POLLIN
  85         if events & POLLOUT:
  86             self.wlist.append(fd)
  87             events &= ~POLLOUT
  88         if events:
  89             self.xlist.append(fd)
  90
  91     def poll(self, timeout):
  92         # XXX workaround a bug in eventlet
  93         # see https://github.com/eventlet/eventlet/pull/25
  94         if timeout == 0 and _using_eventlet_green_select():
  95             timeout = 0.1
  96         if sys.platform == 'win32':
  97             events = self.rlist + self.wlist + self.xlist
  98             if not events:
  99                 return []
 100             if len(events) > winutils.win32event.MAXIMUM_WAIT_OBJECTS:
 101                 raise WindowsError("Cannot handle more than maximum wait"
 102                                    "objects\n")
 103
 104             # win32event.INFINITE timeout is -1
 105             # timeout must be an int number, expressed in ms
 106             if timeout == 0.1:
 107                 timeout = 100
 108             else:
 109                 timeout = int(timeout)
 110
 111             # Wait until any of the events is set to signaled
 112             try:
 113                 retval = winutils.win32event.WaitForMultipleObjects(
 114                     events,
 115                     False,  # Wait all
 116                     timeout)
 117             except winutils.pywintypes.error:
 118                 return [(0, POLLERR)]
 119
 120             if retval == winutils.winerror.WAIT_TIMEOUT:
 121                 return []
 122
 123             if events[retval] in self.rlist:
 124                 revent = POLLIN
 125             elif events[retval] in self.wlist:
 126                 revent = POLLOUT
 127             else:
 128                 revent = POLLERR
 129
 130             return [(events[retval], revent)]
 131         else:
 132             if timeout == -1:
 133                 # epoll uses -1 for infinite timeout, select uses None.
 134                 timeout = None
 135             else:
 136                 timeout = float(timeout) / 1000
 137             rlist, wlist, xlist = select.select(self.rlist,
 138                                                 self.wlist,
 139                                                 self.xlist,
 140                                                 timeout)
 141             events_dict = {}
 142             for fd in rlist:
 143                 events_dict[fd] = events_dict.get(fd, 0) | POLLIN
 144             for fd in wlist:
 145                 events_dict[fd] = events_dict.get(fd, 0) | POLLOUT
 146             for fd in xlist:
 147                 events_dict[fd] = events_dict.get(fd, 0) | (POLLERR |
 148                                                             POLLHUP |
 149                                                             POLLNVAL)
 150             return list(events_dict.items())
 151
 152
 153 SelectPoll = _SelectSelect
 154 # If eventlet/gevent isn't used, we can use select.poll by replacing
 155 # _SelectPoll with select.poll class
 156 # _SelectPoll = select.poll
 157
 158
 159 class Poller(object):
 160     """High-level wrapper around the "poll" system call.
 161
 162     Intended usage is for the program's main loop to go about its business
 163     servicing whatever events it needs to.  Then, when it runs out of immediate
 164     tasks, it calls each subordinate module or object's "wait" function, which
 165     in turn calls one (or more) of the functions Poller.fd_wait(),
 166     Poller.immediate_wake(), and Poller.timer_wait() to register to be awakened
 167     when the appropriate event occurs.  Then the main loop calls
 168     Poller.block(), which blocks until one of the registered events happens."""
 169
 170     def __init__(self):
 171         self.__reset()
 172
 173     def fd_wait(self, fd, events):
 174         """Registers 'fd' as waiting for the specified 'events' (which should
 175         be select.POLLIN or select.POLLOUT or their bitwise-OR).  The following
 176         call to self.block() will wake up when 'fd' becomes ready for one or
 177         more of the requested events.
 178
 179         The event registration is one-shot: only the following call to
 180         self.block() is affected.  The event will need to be re-registered
 181         after self.block() is called if it is to persist.
 182
 183         'fd' may be an integer file descriptor or an object with a fileno()
 184         method that returns an integer file descriptor."""
 185         self.poll.register(fd, events)
 186
 187     def __timer_wait(self, msec):
 188         if self.timeout < 0 or msec < self.timeout:
 189             self.timeout = msec
 190
 191     def timer_wait(self, msec):
 192         """Causes the following call to self.block() to block for no more than
 193         'msec' milliseconds.  If 'msec' is nonpositive, the following call to
 194         self.block() will not block at all.
 195
 196         The timer registration is one-shot: only the following call to
 197         self.block() is affected.  The timer will need to be re-registered
 198         after self.block() is called if it is to persist."""
 199         if msec <= 0:
 200             self.immediate_wake()
 201         else:
 202             self.__timer_wait(msec)
 203
 204     def timer_wait_until(self, msec):
 205         """Causes the following call to self.block() to wake up when the
 206         current time, as returned by ovs.timeval.msec(), reaches 'msec' or
 207         later.  If 'msec' is earlier than the current time, the following call
 208         to self.block() will not block at all.
 209
 210         The timer registration is one-shot: only the following call to
 211         self.block() is affected.  The timer will need to be re-registered
 212         after self.block() is called if it is to persist."""
 213         now = ovs.timeval.msec()
 214         if msec <= now:
 215             self.immediate_wake()
 216         else:
 217             self.__timer_wait(msec - now)
 218
 219     def immediate_wake(self):
 220         """Causes the following call to self.block() to wake up immediately,
 221         without blocking."""
 222         self.timeout = 0
 223
 224     def block(self):
 225         """Blocks until one or more of the events registered with
 226         self.fd_wait() occurs, or until the minimum duration registered with
 227         self.timer_wait() elapses, or not at all if self.immediate_wake() has
 228         been called."""
 229         try:
 230             try:
 231                 events = self.poll.poll(self.timeout)
 232                 self.__log_wakeup(events)
 233             except OSError as e:
 234                 """ On Windows, the select function from poll raises OSError
 235                 exception if the polled array is empty."""
 236                 if e.errno != errno.EINTR:
 237                     vlog.err("poll: %s" % os.strerror(e.errno))
 238             except select.error as e:
 239                 # XXX rate-limit
 240                 error, msg = e
 241                 if error != errno.EINTR:
 242                     vlog.err("poll: %s" % e[1])
 243         finally:
 244             self.__reset()
 245
 246     def __log_wakeup(self, events):
 247         if not events:
 248             vlog.dbg("%d-ms timeout" % self.timeout)
 249         else:
 250             for fd, revents in events:
 251                 if revents != 0:
 252                     s = ""
 253                     if revents & POLLIN:
 254                         s += "[POLLIN]"
 255                     if revents & POLLOUT:
 256                         s += "[POLLOUT]"
 257                     if revents & POLLERR:
 258                         s += "[POLLERR]"
 259                     if revents & POLLHUP:
 260                         s += "[POLLHUP]"
 261                     if revents & POLLNVAL:
 262                         s += "[POLLNVAL]"
 263                     vlog.dbg("%s on fd %d" % (s, fd))
 264
 265     def __reset(self):
 266         self.poll = SelectPoll()
 267         self.timeout = -1
 268
 269
 270 def get_system_poll():
 271     """Returns the original select.poll() object. If select.poll is
 272     monkey patched by eventlet or gevent library, it gets the original
 273     select.poll and returns an object of it using the
 274     eventlet.patcher.original/gevent.monkey.get_original functions.
 275
 276     As a last resort, if there is any exception it returns the
 277     SelectPoll() object.
 278     """
 279     try:
 280         if _using_eventlet_green_select():
 281             _system_poll = eventlet_patcher.original("select").poll
 282         elif gevent_monkey and gevent_monkey.is_object_patched(
 283                 'select', 'poll'):
 284             _system_poll = gevent_monkey.get_original('select', 'poll')
 285         else:
 286             _system_poll = select.poll
 287     except:
 288         _system_poll = SelectPoll
 289
 290     return _system_poll()