def test_write(self):
     p = Popen(["dd", "of=/dev/null", "bs=%d" % self.BUFSIZE],
               stdin=subprocess.PIPE,
               stdout=None,
               stderr=subprocess.PIPE)
     start = monotonic_time()
     total = self.COUNT * self.BUFSIZE
     sent = 0
     with io.open("/dev/zero", "rb") as f:
         while sent < total:
             n = min(total - sent, self.BUFSIZE)
             data = f.read(n)
             if not data:
                 raise RuntimeError("/dev/zero closed?!")
             p.stdin.write(data)
             sent += len(data)
     p.stdin.flush()
     p.stdin.close()
     for _, data in cmdutils.receive(p, 10):
         pass
     elapsed = monotonic_time() - start
     sent_gb = sent / float(1024**3)
     print("%.2fg in %.2f seconds (%.2fg/s)"
           % (sent_gb, elapsed, sent_gb / elapsed), end=" ")
     self.assertEqual(p.returncode, 0)

def stopwatch(message, level=logging.DEBUG,
              log=logging.getLogger('vds.stopwatch')):
    if log.isEnabledFor(level):
        start = vdsm_time.monotonic_time()
        yield
        elapsed = vdsm_time.monotonic_time() - start
        log.log(level, "%s: %.2f seconds", message, elapsed)
    else:
        yield

 def assertMaxDuration(self, maxtime):
     start = time.monotonic_time()
     try:
         yield
     finally:
         elapsed = time.monotonic_time() - start
         if maxtime < elapsed:
             self.fail("Operation was too slow %.2fs > %.2fs" %
                       (elapsed, maxtime))

 def _work():
     invocations[0] += 1
     invocations[1] = monotonic_time()
     if invocations[0] == BLOCK_AT:
         # must be > (PERIOD * TIMES) ~= forever
         time.sleep(10 * PERIOD * TIMES)
     executions[0] += 1
     executions[1] = monotonic_time()
     if invocations[0] == TIMES:
         done.set()

def _wait_for_link_up(devname, timeout):
    """
    Waiting for link-up, no longer than the specified timeout period.
    The time waited (in seconds) is returned.
    """
    if timeout > 0 and not iface_obj(devname).is_oper_up():
        time_start = monotonic_time()
        with waitfor.waitfor_linkup(devname, timeout=timeout):
            pass
        return monotonic_time() - time_start
    return 0

    def test_timeout_not_triggered(self):
        time_start = monotonic_time()
        with monitor.Monitor(timeout=self.TIMEOUT) as mon:
            dummy = Dummy()
            dummy.create()
            dummy.remove()

            for event in mon:
                break

        self.assertTrue((monotonic_time() - time_start) <= self.TIMEOUT)
        self.assertTrue(mon.is_stopped())

 def _serveRequest(self, ctx, req):
     start_time = monotonic_time()
     response = self._handle_request(req, ctx)
     error = getattr(response, "error", None)
     if error is None:
         response_log = "succeeded"
     else:
         response_log = "failed (error %s)" % (error.code,)
     self.log.info("RPC call %s %s in %.2f seconds",
                   req.method, response_log, monotonic_time() - start_time)
     if response is not None:
         ctx.requestDone(response)

def _wait_for_socket(sock, timeout):
    start = monotonic_time()
    elapsed = 0.0

    while elapsed < timeout:
        if os.path.exists(sock):
            log.debug("Waited for socket %.3f seconds", elapsed)
            return True
        # Socket is usually availble after 20 milliseconds.
        time.sleep(0.02)
        elapsed = monotonic_time() - start

    return False

def _wait(p, deadline=None):
    """
    Wait until process terminates, or if deadline is specified,
    `common.time.monotonic_time` exceeds deadline.

    Raises:
        `cmdutils.TimeoutExpired` if process did not terminate within
            deadline.
    """
    log.debug("Waiting for process (pid=%d)", p.pid)
    if deadline is None:
        p.wait()
    else:
        # We need to wait until deadline, Popen.wait() does not support
        # timeout. Python 3 is using busy wait in this case with a timeout of
        # 0.0005 seocnds. In vdsm we cannot allow such busy loops, and we don't
        # have a need to support very exact wait time. This loop uses
        # exponential backoff to detect termination quickly if the process
        # terminates quickly, and avoid busy loop if the process is stuck for
        # long time. Timeout will double from 0.0078125 to 1.0, and then
        # continue at 1.0 seconds, until deadline is reached.
        timeout = 1.0 / 256
        while p.poll() is None:
            remaining = deadline - monotonic_time()
            if remaining <= 0:
                raise TimeoutExpired(p.pid)
            time.sleep(min(timeout, remaining))
            if timeout < 1.0:
                timeout *= 2
    log.debug("Process (pid=%d) terminated", p.pid)

    def _scan(self):
        with closing(select.epoll()) as epoll:
            with _monitoring_socket(self._queue, self._groups, epoll) as sock:
                with _pipetrick(epoll) as self._pipetrick:
                    self._scanning_started.set()
                    while True:
                        if self._timeout:
                            timeout = self._end_time - monotonic_time()
                            # timeout expired
                            if timeout <= 0:
                                self._scanning_stopped.set()
                                self._queue.put(_TIMEOUT_FLAG)
                                break
                        else:
                            timeout = -1

                        events = uninterruptible_poll(epoll.poll,
                                                      timeout=timeout)
                        # poll timeouted
                        if len(events) == 0:
                            self._scanning_stopped.set()
                            self._queue.put(_TIMEOUT_FLAG)
                            break
                        # stopped by pipetrick
                        elif (self._pipetrick[0], select.POLLIN) in events:
                            uninterruptible(os.read, self._pipetrick[0], 1)
                            self._queue.put(_STOP_FLAG)
                            break

                        libnl.nl_recvmsgs_default(sock)

 def tick(self):
     now = monotonic_time()
     result = self._result(now)
     self._counter = (self._counter + 1) % self._interval
     if result:
         self._last_time = now
     return result

 def _attempt_log_stats(self):
     self._counter += 1
     if monotonic_time() > self._next_report:
         self.log.info('%s requests processed during %s seconds',
                       self._counter, self._timeout)
         self._next_report += self._timeout
         self._counter = 0

    def receive(self, timeout=None):
        """
        Receiving data from the command can raise OSError
        exceptions as described in read(2).
        """
        if timeout is None:
            poll_remaining = -1
        else:
            endtime = vdsm_time.monotonic_time() + timeout

        while not self.closed:
            if timeout is not None:
                poll_remaining = endtime - vdsm_time.monotonic_time()
                if poll_remaining <= 0:
                    break

            self._poll_timeout(poll_remaining)

 def test_read(self):
     p = Popen(["dd", "if=/dev/zero", "bs=%d" % self.BUFSIZE,
                "count=%d" % self.COUNT],
               stdin=None,
               stdout=subprocess.PIPE,
               stderr=subprocess.PIPE)
     start = monotonic_time()
     received = 0
     for src, data in cmdutils.receive(p, bufsize=self.BUFSIZE):
         if src == cmdutils.OUT:
             received += len(data)
     elapsed = monotonic_time() - start
     received_gb = received / float(1024**3)
     print("%.2fg in %.2f seconds (%.2fg/s)"
           % (received_gb, elapsed, received_gb / elapsed), end=" ")
     self.assertEqual(received, self.COUNT * self.BUFSIZE)
     self.assertEqual(p.returncode, 0)

def _wait_for_for_all_devices_up(links):
    timeout = monotonic_time() + _ALL_DEVICES_UP_TIMEOUT
    down_links = _get_links_with_state_down(links)

    # TODO: use netlink monitor here might be more elegant (not available in
    # TODO: 3.5)
    while down_links and monotonic_time() < timeout:
        logging.debug("waiting for %s to be up.", down_links)
        time.sleep(1)
        down_links = _get_links_with_state_down(links)

    if down_links:
        logging.warning("Not all devices are up. VDSM might restore them "
                        "although they were not changed since they were "
                        "persisted.")
    else:
        logging.debug("All devices are up.")

 def __init__(self, bridge, timeout, cif, threadFactory=None):
     self._bridge = bridge
     self._cif = cif
     self._workQueue = queue.Queue()
     self._threadFactory = threadFactory
     self._timeout = timeout
     self._next_report = monotonic_time() + self._timeout
     self._counter = 0

 def runnable(self):
     if not self._vm.isDomainReadyForCommands():
         return False
     last_failure = self._qga_poller.last_failure(self._vm.id)
     if last_failure is not None and \
             (monotonic_time() - last_failure) < _THROTTLING_INTERVAL:
         return False
     return True

 def check_estimate(self, filename, compat):
     start = time.monotonic_time()
     estimate = qcow2.estimate_size(filename)
     estimate_time = time.monotonic_time() - start
     start = time.monotonic_time()
     actual = converted_size(filename, compat)
     convert_time = time.monotonic_time() - start
     original_size = os.stat(filename).st_size
     error_pct = 100 * float(estimate - actual) / original_size
     print('estimate=%d, '
           'actual=%s, '
           'error_pct=%.2f%%, '
           'estimate_time=%.2f, '
           'convert_time=%.2f'
           % (estimate, actual, error_pct, estimate_time, convert_time),
           end=" ")
     self.assertGreaterEqual(estimate, actual)
     self.assertGreaterEqual(0.1, error_pct)

def safe_poll(mp_connection, timeout):
    """
    This is a workaround until we get the PEP-475 fix for EINTR.  It
    ensures that a multiprocessing.connection.poll() will not return
    before the timeout due to an interruption.

    Returns True if there is any data to read from the pipe or if the
    pipe was closed.  Returns False if the timeout expired.
    """
    deadline = time.monotonic_time() + timeout
    remaining = timeout

    while not mp_connection.poll(remaining):
        remaining = deadline - time.monotonic_time()
        if remaining <= 0:
            return False

    return True

 def _wait_for_state(self, state, deadline):
     while self._state != state:
         if deadline is not None:
             now = time.monotonic_time()
             if now >= deadline:
                 raise Timeout("Timeout waiting for barrier")
             self._cond.wait(deadline - now)
         else:
             self._cond.wait()