限制python多处理中的总CPU使用率

Question

我使用multiprocessing.Pool.imap在Windows 7上使用Python 2.7并行运行许多独立作业.使用默认设置,我的总CPU使用率固定为100%,由Windows任务管理器测量.这使得我的代码在后台运行时无法执行任何其他工作.

我已经尝试将进程数限制为CPU数减1,如如何限制Python使用的处理器数量中所述:

pool = Pool(processes=max(multiprocessing.cpu_count()-1, 1)
for p in pool.imap(func, iterable):
     ...

这确实减少了正在运行的进程总数.但是,每个过程只需要更多的周期来弥补它.所以我的总CPU使用率仍然固定为100%.

有没有办法直接限制总CPU使用率 - 不仅仅是进程数量 - 或者失败,是否有任何解决方法？

Answer 1

解决方案取决于您想要做什么.以下是一些选项:

流程优先级较低

你可以nice使用子流程.这样,虽然它们仍会占用100%的CPU,但当您启动其他应用程序时,操作系统会优先考虑其他应用程序.如果您想在笔记本电脑的背景上运行工作密集型计算而不关心CPU风扇一直运行,那么设置好的值psutils就是您的解决方案.此脚本是一个测试脚本,它在所有核心上运行足够长的时间,以便您可以看到它的行为方式.

from multiprocessing import Pool, cpu_count
import math
import psutil
import os

def f(i):
    return math.sqrt(i)

def limit_cpu():
    "is called at every process start"
    p = psutil.Process(os.getpid())
    # set to lowest priority, this is windows only, on Unix use ps.nice(19)
    p.nice(psutil.BELOW_NORMAL_PRIORITY_CLASS)

if __name__ == '__main__':
    # start "number of cores" processes
    pool = Pool(None, limit_cpu)
    for p in pool.imap(f, range(10**8)):
        pass

诀窍在于limit_cpu每个进程的开始运行(请参阅initializerdoc中的argment).虽然Unix的级别为-19(最高prio)到19(最低prio),但Windows有一些不同的优先级.BELOW_NORMAL_PRIORITY_CLASS可能最符合您的要求,还有IDLE_PRIORITY_CLASS一个说Windows只在系统空闲时运行您的进程.

如果在任务管理器中切换到详细模式并右键单击该过程,则可以查看优先级:

较少的进程数

虽然您已拒绝此选项,但它仍然可能是一个不错的选择:假设您将子pool = Pool(max(cpu_count()//2, 1))进程的数量限制为cpu核心的一半,然后操作系统最初在一半的cpu核心上运行这些进程,而其他进程保持空闲或只运行另一个目前正在运行 在很短的时间之后,操作系统重新安排进程并可能将它们移动到其他cpu内核等.作为基于Unix的系统的Windows都以这种方式运行.

Windows:在4个核心上运行2个进程:

OSX:在8个核心上运行4个进程:

您可以看到两个操作系统都在核心之间平衡了进程,尽管不均匀,因此您仍然可以看到一些核心比其他核心更高的核心.

睡觉

如果你绝对想要确定,你的进程永远不会吃100%的某个核心(例如,如果你想防止cpu风扇上升),那么你可以在你的处理函数中运行sleep:

from time import sleep

def f(i):
    sleep(0.01)
    return math.sqrt(i)

这使得OS 0.01为每个计算"安排"您的过程几秒钟,并为其他应用程序腾出空间.如果没有其他应用程序,则cpu核心处于空闲状态,因此它永远不会达到100%.您需要使用不同的睡眠持续时间,它也会因您运行它而在计算机之间变化.如果你想让它变得非常复杂,你可以根据cpu_times()报告调整睡眠.

Answer 2

在操作系统层面

您可以使用nice为单个命令设置优先级。你也可以用nice启动一个python脚本。（以下来自：http://blog.scoutapp.com/articles/2014/11/04/restricting-process-cpu-usage-using-nice-cpulimit-and-cgroups）

好的

Nice 命令调整进程的优先级，以降低其运行频率。当您需要将 CPU 密集型任务作为后台或批处理作业运行时，这非常有用。友好度级别范围从 -20（最有利的调度）到 19（最不有利的调度）。Linux 上的进程默认以 0 的良好度启动。Nice 命令（不带任何附加参数）将启动一个 Niceness 为 10 的进程。在该级别，调度程序会将其视为优先级较低的任务，并为其提供较少的 CPU 资源。启动两个 matho-primes 任务，其中一个带有 Nice 和一个没有：

nice matho-primes 0 9999999999 > /dev/null &matho-primes 0 9999999999 > /dev/null &
matho-primes 0 9999999999 > /dev/null &

现在跑到上面。

作为 Python 中的函数

另一种方法是使用 psutils 检查过去一分钟的 CPU 平均负载，然后让您的线程检查 CPU 平均负载，如果低于指定的 CPU 负载目标，则假脱机另一个线程，如果低于指定的 CPU 负载目标，则休眠或终止该线程。高于 CPU 负载目标。当您使用计算机时，这不会妨碍您，但会保持恒定的 CPU 负载。

# Import Python modules
import time
import os
import multiprocessing
import psutil
import math
from random import randint

# Main task function
def main_process(item_queue, args_array):
    # Go through each link in the array passed in.
    while not item_queue.empty():
        # Get the next item in the queue
        item = item_queue.get()
        # Create a random number to simulate threads that
        # are not all going to be the same
        randomizer = randint(100, 100000)
        for i in range(randomizer):
            algo_seed = math.sqrt(math.sqrt(i * randomizer) % randomizer)
        # Check if the thread should continue based on current load balance
        if spool_down_load_balance():
            print "Process " + str(os.getpid()) + " saying goodnight..."
            break

# This function will build a queue and
def start_thread_process(queue_pile, args_array):
    # Create a Queue to hold link pile and share between threads
    item_queue = multiprocessing.Queue()
    # Put all the initial items into the queue
    for item in queue_pile:
        item_queue.put(item)
    # Append the load balancer thread to the loop
    load_balance_process = multiprocessing.Process(target=spool_up_load_balance, args=(item_queue, args_array))
    # Loop through and start all processes
    load_balance_process.start()
    # This .join() function prevents the script from progressing further.
    load_balance_process.join()

# Spool down the thread balance when load is too high
def spool_down_load_balance():
    # Get the count of CPU cores
    core_count = psutil.cpu_count()
    # Calulate the short term load average of past minute
    one_minute_load_average = os.getloadavg()[0] / core_count
    # If load balance above the max return True to kill the process
    if one_minute_load_average > args_array['cpu_target']:
        print "-Unacceptable load balance detected. Killing process " + str(os.getpid()) + "..."
        return True

# Load balancer thread function
def spool_up_load_balance(item_queue, args_array):

    print "[Starting load balancer...]"
    # Get the count of CPU cores
    core_count = psutil.cpu_count()
    # While there is still links in queue
    while not item_queue.empty():
        print "[Calculating load balance...]"
        # Check the 1 minute average CPU load balance
        # returns 1,5,15 minute load averages
        one_minute_load_average = os.getloadavg()[0] / core_count
        # If the load average much less than target, start a group of new threads
        if one_minute_load_average < args_array['cpu_target'] / 2:
            # Print message and log that load balancer is starting another thread
            print "Starting another thread group due to low CPU load balance of: " + str(one_minute_load_average * 100) + "%"
            time.sleep(5)
            # Start another group of threads
            for i in range(3):
                start_new_thread = multiprocessing.Process(target=main_process,args=(item_queue, args_array))
                start_new_thread.start()
            # Allow the added threads to have an impact on the CPU balance
            # before checking the one minute average again
            time.sleep(20)

        # If load average less than target start single thread
        elif one_minute_load_average < args_array['cpu_target']:
            # Print message and log that load balancer is starting another thread
            print "Starting another single thread due to low CPU load balance of: " + str(one_minute_load_average * 100) + "%"
            # Start another thread
            start_new_thread = multiprocessing.Process(target=main_process,args=(item_queue, args_array))
            start_new_thread.start()
            # Allow the added threads to have an impact on the CPU balance
            # before checking the one minute average again
            time.sleep(20)

        else:
            # Print CPU load balance
            print "Reporting stable CPU load balance: " + str(one_minute_load_average * 100) + "%"
            # Sleep for another minute while
            time.sleep(20)

if __name__=="__main__":

    # Set the queue size
    queue_size = 10000
    # Define an arguments array to pass around all the values
    args_array = {
        # Set some initial CPU load values as a CPU usage goal
        "cpu_target" : 0.60,
        # When CPU load is significantly low, start this number
        # of threads
        "thread_group_size" : 3
    }

    # Create an array of fixed length to act as queue
    queue_pile = list(range(queue_size))
    # Set main process start time
    start_time = time.time()
    # Start the main process
    start_thread_process(queue_pile, args_array)
    print '[Finished processing the entire queue! Time consuming:{0} Time Finished: {1}]'.format(time.time() - start_time, time.strftime("%c"))