python类getpriority()的实例源码

def test_set_get_priority(self):

        base = os.getpriority(os.PRIO_PROCESS, os.getpid())
        os.setpriority(os.PRIO_PROCESS, os.getpid(), base + 1)
        try:
            new_prio = os.getpriority(os.PRIO_PROCESS, os.getpid())
            if base >= 19 and new_prio <= 19:
                raise unittest.SkipTest(
      "unable to reliably test setpriority at current nice level of %s" % base)
            else:
                self.assertEqual(new_prio, base + 1)
        finally:
            try:
                os.setpriority(os.PRIO_PROCESS, os.getpid(), base)
            except OSError as err:
                if err.errno != errno.EACCES:
                    raise

def test_set_get_priority(self):

        base = os.getpriority(os.PRIO_PROCESS, os.getpid())
        os.setpriority(os.PRIO_PROCESS, os.getpid(), base + 1)
        try:
            new_prio = os.getpriority(os.PRIO_PROCESS, os.getpid())
            if base >= 19 and new_prio <= 19:
                raise unittest.SkipTest(
      "unable to reliably test setpriority at current nice level of %s" % base)
            else:
                self.assertEqual(new_prio, base + 1)
        finally:
            try:
                os.setpriority(os.PRIO_PROCESS, os.getpid(), base)
            except OSError as err:
                if err.errno != errno.EACCES:
                    raise

def test_set_get_priority(self):

        base = os.getpriority(os.PRIO_PROCESS, os.getpid())
        os.setpriority(os.PRIO_PROCESS, os.getpid(), base + 1)
        try:
            new_prio = os.getpriority(os.PRIO_PROCESS, os.getpid())
            if base >= 19 and new_prio <= 19:
                raise unittest.SkipTest(
      "unable to reliably test setpriority at current nice level of %s" % base)
            else:
                self.assertEqual(new_prio, base + 1)
        finally:
            try:
                os.setpriority(os.PRIO_PROCESS, os.getpid(), base)
            except OSError as err:
                if err.errno != errno.EACCES:
                    raise

def test_nice(self):
        p = psutil.Process()
        self.assertRaises(TypeError, p.nice, "str")
        if WINDOWS:
            try:
                init = p.nice()
                if sys.version_info > (3, 4):
                    self.assertIsInstance(init, enum.IntEnum)
                else:
                    self.assertIsInstance(init, int)
                self.assertEqual(init, psutil.NORMAL_PRIORITY_CLASS)
                p.nice(psutil.HIGH_PRIORITY_CLASS)
                self.assertEqual(p.nice(), psutil.HIGH_PRIORITY_CLASS)
                p.nice(psutil.NORMAL_PRIORITY_CLASS)
                self.assertEqual(p.nice(), psutil.NORMAL_PRIORITY_CLASS)
            finally:
                p.nice(psutil.NORMAL_PRIORITY_CLASS)
        else:
            first_nice = p.nice()
            try:
                if hasattr(os, "getpriority"):
                    self.assertEqual(
                        os.getpriority(os.PRIO_PROCESS, os.getpid()), p.nice())
                p.nice(1)
                self.assertEqual(p.nice(), 1)
                if hasattr(os, "getpriority"):
                    self.assertEqual(
                        os.getpriority(os.PRIO_PROCESS, os.getpid()), p.nice())
                # XXX - going back to previous nice value raises
                # AccessDenied on OSX
                if not OSX:
                    p.nice(0)
                    self.assertEqual(p.nice(), 0)
            except psutil.AccessDenied:
                pass
            finally:
                try:
                    p.nice(first_nice)
                except psutil.AccessDenied:
                    pass

def set_process_priority(nice: bool=True, ionice: bool=False,
                         cpulimit: int=0) -> bool:
    """Set process name and cpu priority."""
    w = " may delay I/O Operations, not recommended on user-facing GUI!."
    try:
        if nice:
            _old = os.getpriority(os.PRIO_PROCESS, 0)
            os.nice(19)  # smooth cpu priority
            log.debug(f"Process CPUs Priority set: from {_old} to 19.")
        elif ionice and which("ionice"):
            log.warning("ionice" + w)
            cmnd = f"{which('ionice')} --ignore --class 3 --pid {os.getpid()}"
            call(cmnd, shell=True)  # I/O nice,should work on Linux and Os X
            log.debug(f"Process PID {os.getpid()} I/O Priority set to: {cmnd}")
        elif cpulimit and which("cpulimit"):
            log.warning("cpulimit" + w)
            log.debug("Launching 1 background 'cpulimit' child subprocess...")
            cpulimit = int(cpulimit if cpulimit > 4 else 5)  # makes 5 the min.
            command = "{0} --include-children --pid={1} --limit={2}".format(
                which("cpulimit"), os.getpid(), cpulimit)
            proces = Popen(command, shell=True)  # This launch a subprocess.
            atexit.register(proces.kill)  # Force Kill subprocess at exit.
            log.debug(f"Process CPUs Max Limits capped to: {command}.")
    except Exception as error:
        log.warning(error)
        return False  # this may fail on windows and its normal, so be silent.
    else:
        return True

def test_set_process_priority(self):
        old_nice = os.getpriority(os.PRIO_PROCESS, 0)
        prio_set = set_process_priority(ionice=True)
        new_nice = os.getpriority(os.PRIO_PROCESS, 0)
        self.assertEqual(old_nice, 0)  # a == b
        self.assertTrue(isinstance(prio_set, bool))  # bool(x) is True
        self.assertTrue(prio_set)  # bool(x) is True
        self.assertEqual(new_nice, 19)  # a == b