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