def run(self):
        """Run the proton event/timer loop."""
        LOG.debug("Starting Proton thread, container=%s",
                  self._container.name)

        while not self._shutdown:

            readfds = [self._requests]
            writefds = []
            deadline = self._scheduler._next_deadline

            pyngus_conn = self._connection and self._connection.pyngus_conn
            if pyngus_conn and self._connection.socket:
                if pyngus_conn.needs_input:
                    readfds.append(self._connection)
                if pyngus_conn.has_output:
                    writefds.append(self._connection)
                if pyngus_conn.deadline:
                    deadline = (pyngus_conn.deadline if not deadline else
                                min(deadline, pyngus_conn.deadline))

            # force select to return in time to service the next expiring timer
            if deadline:
                _now = now()
                timeout = 0 if deadline <= _now else (deadline - _now)
            else:
                timeout = None

            # and now we wait...
            try:
                select.select(readfds, writefds, [], timeout)
            except select.error as serror:
                if serror[0] == errno.EINTR:
                    LOG.warning(_LW("ignoring interrupt from select(): %s"),
                                str(serror))
                    continue
                raise  # assuming fatal...

            # Ignore the select return value - simply poll the socket for I/O.
            # Testing shows that polling improves latency over checking the
            # lists returned by select()
            self._requests.process_requests()
            self._connection.read_socket()
            if pyngus_conn and pyngus_conn.deadline:
                _now = now()
                if pyngus_conn.deadline <= _now:
                    pyngus_conn.process(_now)
            self._connection.write_socket()

            self._scheduler._process()  # run any deferred requests

        LOG.info(_LI("eventloop thread exiting, container=%s"),
                 self._container.name)

    def run_periodic_tasks(self, context, raise_on_error=False):
        """Tasks to be run at a periodic interval."""
        idle_for = DEFAULT_INTERVAL
        for task_name, task in self._periodic_tasks:
            if (task._periodic_external_ok and not
               self.conf.run_external_periodic_tasks):
                continue
            full_task_name = '.'.join([self.__class__.__name__, task_name])

            spacing = self._periodic_spacing[task_name]
            last_run = self._periodic_last_run[task_name]

            # Check if due, if not skip
            idle_for = min(idle_for, spacing)
            if last_run is not None:
                delta = last_run + spacing - now()
                if delta > 0:
                    idle_for = min(idle_for, delta)
                    continue

            LOG.debug("Running periodic task %(full_task_name)s",
                      {"full_task_name": full_task_name})
            self._periodic_last_run[task_name] = _nearest_boundary(
                last_run, spacing)

            try:
                task(self, context)
            except Exception:
                if raise_on_error:
                    raise
                LOG.exception(_LE("Error during %(full_task_name)s"),
                              {"full_task_name": full_task_name})
            time.sleep(0)

        return idle_for

 def stop(self):
     """Stops the watch."""
     if self._state == self._STOPPED:
         return self
     if self._state != self._STARTED:
         raise RuntimeError("Can not stop a stopwatch that has not been"
                            " started")
     self._stopped_at = now()
     self._state = self._STOPPED
     return self

 def write_socket(self):
     """Called to write to the socket."""
     if self.socket:
         try:
             pyngus.write_socket_output(self.pyngus_conn, self.socket)
             self.pyngus_conn.process(now())
         except (socket.timeout, socket.error) as e:
             # pyngus handles EAGAIN/EWOULDBLOCK and EINTER
             self.pyngus_conn.close_output()
             self.pyngus_conn.close_input()
             self._handler.socket_error(str(e))

 def elapsed(self, maximum=None):
     """Returns how many seconds have elapsed."""
     if self._state not in (self._STARTED, self._STOPPED):
         raise RuntimeError("Can not get the elapsed time of a stopwatch"
                            " if it has not been started/stopped")
     if self._state == self._STOPPED:
         elapsed = self._delta_seconds(self._started_at, self._stopped_at)
     else:
         elapsed = self._delta_seconds(self._started_at, now())
     if maximum is not None and elapsed > maximum:
         elapsed = max(0.0, maximum)
     return elapsed

    def start(self):
        """Starts the watch (if not already started).

        NOTE(harlowja): resets any splits previously captured (if any).
        """
        if self._state == self._STARTED:
            return self
        self._started_at = now()
        self._stopped_at = None
        self._state = self._STARTED
        self._splits = []
        return self

 def _process(self):
     """Invoke all expired callables."""
     if self._deadlines:
         _now = now()
         try:
             while self._deadlines[0] <= _now:
                 deadline = heapq.heappop(self._deadlines)
                 callbacks = self._callbacks[deadline]
                 del self._callbacks[deadline]
                 for cb in callbacks:
                     cb.callback and cb.callback()
         except IndexError:
             pass

 def _get_delay(self, max_delay=None):
     """Get the delay in milliseconds until the next callable needs to be
     run, or 'max_delay' if no outstanding callables or the delay to the
     next callable is > 'max_delay'.
     """
     due = self._deadlines[0] if self._deadlines else None
     if due is None:
         return max_delay
     _now = now()
     if due <= _now:
         return 0
     else:
         return min(due - _now, max_delay) if max_delay else due - _now

    def decorator(f):
        # Test for old style invocation
        if 'ticks_between_runs' in kwargs:
            raise InvalidPeriodicTaskArg(arg='ticks_between_runs')

        # Control if run at all
        f._periodic_task = True
        f._periodic_external_ok = kwargs.pop('external_process_ok', False)
        f._periodic_enabled = kwargs.pop('enabled', True)
        f._periodic_name = kwargs.pop('name', f.__name__)

        # Control frequency
        f._periodic_spacing = kwargs.pop('spacing', 0)
        f._periodic_immediate = kwargs.pop('run_immediately', False)
        if f._periodic_immediate:
            f._periodic_last_run = None
        else:
            f._periodic_last_run = now()
        return f

def _nearest_boundary(last_run, spacing):
    """Find the nearest boundary in the past.

    The boundary is a multiple of the spacing with the last run as an offset.

    Eg if last run was 10 and spacing was 7, the new last run could be: 17, 24,
    31, 38...

    0% to 5% of the spacing value will be added to this value to ensure tasks
    do not synchronize. This jitter is rounded to the nearest second, this
    means that spacings smaller than 20 seconds will not have jitter.
    """
    current_time = now()
    if last_run is None:
        return current_time
    delta = current_time - last_run
    offset = delta % spacing
    # Add up to 5% jitter
    jitter = int(spacing * (random.random() / 20))
    return current_time - offset + jitter

def compute_timeout(offset):
    # minimize the timer granularity to one second so we don't have to track
    # too many timers
    return math.ceil(now() + offset)

 def start(self):
     self.started_at = now()
     self.stopped_at = None

 def __exit__(self, exc_type, exc_value, exc_tb):
     self.stopped_at = now()

 def elapsed(self):
     if self.stopped_at is not None:
         end_time = self.stopped_at
     else:
         end_time = now()
     return max(0.0, end_time - self.started_at)