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 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         events_dict = {}
  79         for fd in rlist:
  80             events_dict[fd] = events_dict.get(fd, 0) | POLLIN
  81         for fd in wlist:
  82             events_dict[fd] = events_dict.get(fd, 0) | POLLOUT
  83         for fd in xlist:
  84             events_dict[fd] = events_dict.get(fd, 0) | (POLLERR |
  85                                                         POLLHUP |
  86                                                         POLLNVAL)
  87         return events_dict.items()
  88
  89
  90 SelectPoll = _SelectSelect
  91 # If eventlet/gevent isn't used, we can use select.poll by replacing
  92 # _SelectPoll with select.poll class
  93 # _SelectPoll = select.poll
  94
  95
  96 class Poller(object):
  97     """High-level wrapper around the "poll" system call.
  98
  99     Intended usage is for the program's main loop to go about its business
 100     servicing whatever events it needs to.  Then, when it runs out of immediate
 101     tasks, it calls each subordinate module or object's "wait" function, which
 102     in turn calls one (or more) of the functions Poller.fd_wait(),
 103     Poller.immediate_wake(), and Poller.timer_wait() to register to be awakened
 104     when the appropriate event occurs.  Then the main loop calls
 105     Poller.block(), which blocks until one of the registered events happens."""
 106
 107     def __init__(self):
 108         self.__reset()
 109
 110     def fd_wait(self, fd, events):
 111         """Registers 'fd' as waiting for the specified 'events' (which should
 112         be select.POLLIN or select.POLLOUT or their bitwise-OR).  The following
 113         call to self.block() will wake up when 'fd' becomes ready for one or
 114         more of the requested events.
 115
 116         The event registration is one-shot: only the following call to
 117         self.block() is affected.  The event will need to be re-registered
 118         after self.block() is called if it is to persist.
 119
 120         'fd' may be an integer file descriptor or an object with a fileno()
 121         method that returns an integer file descriptor."""
 122         self.poll.register(fd, events)
 123
 124     def __timer_wait(self, msec):
 125         if self.timeout < 0 or msec < self.timeout:
 126             self.timeout = msec
 127
 128     def timer_wait(self, msec):
 129         """Causes the following call to self.block() to block for no more than
 130         'msec' milliseconds.  If 'msec' is nonpositive, the following call to
 131         self.block() will not block at all.
 132
 133         The timer registration is one-shot: only the following call to
 134         self.block() is affected.  The timer will need to be re-registered
 135         after self.block() is called if it is to persist."""
 136         if msec <= 0:
 137             self.immediate_wake()
 138         else:
 139             self.__timer_wait(msec)
 140
 141     def timer_wait_until(self, msec):
 142         """Causes the following call to self.block() to wake up when the
 143         current time, as returned by ovs.timeval.msec(), reaches 'msec' or
 144         later.  If 'msec' is earlier than the current time, the following call
 145         to self.block() will not block at all.
 146
 147         The timer registration is one-shot: only the following call to
 148         self.block() is affected.  The timer will need to be re-registered
 149         after self.block() is called if it is to persist."""
 150         now = ovs.timeval.msec()
 151         if msec <= now:
 152             self.immediate_wake()
 153         else:
 154             self.__timer_wait(msec - now)
 155
 156     def immediate_wake(self):
 157         """Causes the following call to self.block() to wake up immediately,
 158         without blocking."""
 159         self.timeout = 0
 160
 161     def block(self):
 162         """Blocks until one or more of the events registered with
 163         self.fd_wait() occurs, or until the minimum duration registered with
 164         self.timer_wait() elapses, or not at all if self.immediate_wake() has
 165         been called."""
 166         try:
 167             try:
 168                 events = self.poll.poll(self.timeout)
 169                 self.__log_wakeup(events)
 170             except select.error, e:
 171                 # XXX rate-limit
 172                 error, msg = e
 173                 if error != errno.EINTR:
 174                     vlog.err("poll: %s" % e[1])
 175         finally:
 176             self.__reset()
 177
 178     def __log_wakeup(self, events):
 179         if not events:
 180             vlog.dbg("%d-ms timeout" % self.timeout)
 181         else:
 182             for fd, revents in events:
 183                 if revents != 0:
 184                     s = ""
 185                     if revents & POLLIN:
 186                         s += "[POLLIN]"
 187                     if revents & POLLOUT:
 188                         s += "[POLLOUT]"
 189                     if revents & POLLERR:
 190                         s += "[POLLERR]"
 191                     if revents & POLLHUP:
 192                         s += "[POLLHUP]"
 193                     if revents & POLLNVAL:
 194                         s += "[POLLNVAL]"
 195                     vlog.dbg("%s on fd %d" % (s, fd))
 196
 197     def __reset(self):
 198         self.poll = SelectPoll()
 199         self.timeout = -1