使用 Python 在 Windows 上实现并发/并行

Question

我开发了一个简单的程序来解决八皇后问题。现在我想用不同的元参数做更多的测试，所以我想让它快点。我经历了几次分析迭代，能够显着缩短运行时间，但我认为只有部分计算同时进行才能使其更快。我尝试使用multiprocessing 和concurrent.futures 模块，但它并没有大大改善运行时间，在某些情况下甚至减慢了执行速度。那只是提供一些上下文。

我能够提出类似的代码结构，其中顺序版本优于并发。

import numpy as np
import concurrent.futures
import math
import time
import multiprocessing

def is_prime(n):
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def generate_data(seed):
    np.random.seed(seed)
    numbers = []
    for _ in range(5000):
        nbr = np.random.randint(50000, 100000)
        numbers.append(nbr)
    return numbers

def run_test_concurrent(numbers):
    print("Concurrent test")
    start_tm = time.time()
    chunk = len(numbers)//3
    primes = None
    with concurrent.futures.ProcessPoolExecutor(max_workers=3) as pool:
        primes = list(pool.map(is_prime, numbers, chunksize=chunk))
    print("Time: {:.6f}".format(time.time() - start_tm))
    print("Number of primes: {}\n".format(np.sum(primes)))


def run_test_sequential(numbers):
    print("Sequential test")
    start_tm = time.time()
    primes = [is_prime(nbr) for nbr in numbers]
    print("Time: {:.6f}".format(time.time() - start_tm))
    print("Number of primes: {}\n".format(np.sum(primes)))


def run_test_multiprocessing(numbers):
    print("Multiprocessing test")
    start_tm = time.time()
    chunk = len(numbers)//3
    primes = None
    with multiprocessing.Pool(processes=3) as pool:
        primes = list(pool.map(is_prime, numbers, chunksize=chunk))
    print("Time: {:.6f}".format(time.time() - start_tm))
    print("Number of primes: {}\n".format(np.sum(primes)))


def main():
    nbr_trails = 5
    for trail in range(nbr_trails):
        numbers = generate_data(trail*10)
        run_test_concurrent(numbers)
        run_test_sequential(numbers)
        run_test_multiprocessing(numbers)
        print("--\n")


if __name__ == '__main__':
    main()

当我在我的机器上运行它时 - Windows 7，四核 Intel Core i5，我得到以下输出：

Concurrent test
Time: 2.006006
Number of primes: 431

Sequential test
Time: 0.010000
Number of primes: 431

Multiprocessing test
Time: 1.412003
Number of primes: 431
--

Concurrent test
Time: 1.302003
Number of primes: 447

Sequential test
Time: 0.010000
Number of primes: 447

Multiprocessing test
Time: 1.252003
Number of primes: 447
--

Concurrent test
Time: 1.280002
Number of primes: 446

Sequential test
Time: 0.010000
Number of primes: 446

Multiprocessing test
Time: 1.250002
Number of primes: 446
--

Concurrent test
Time: 1.260002
Number of primes: 446

Sequential test
Time: 0.010000
Number of primes: 446

Multiprocessing test
Time: 1.250002
Number of primes: 446
--

Concurrent test
Time: 1.282003
Number of primes: 473

Sequential test
Time: 0.010000
Number of primes: 473

Multiprocessing test
Time: 1.260002
Number of primes: 473
--

我的问题是我是否可以通过在 Windows 上与Python 3.6.4 |Anaconda, Inc.| 同时运行它来使其更快。我在 SO (Why is creating a new process more expensive on Windows than Linux?) 上读到，在 Windows 上创建新进程非常昂贵。有什么办法可以加快速度吗？我错过了一些明显的东西吗？

我也尝试只创建一次Pool，但似乎没有太大帮助。

---

编辑：

原始代码结构看起来或多或少像：

我的代码或多或少是这样的结构：

class Foo(object):

    def g() -> int:
        # function performing simple calculations
        # single function call is fast (~500 ms)
        pass


def run(self):
    nbr_processes = multiprocessing.cpu_count() - 1

    with multiprocessing.Pool(processes=nbr_processes) as pool:
        foos = get_initial_foos()

        solution_found = False
        while not solution_found:
            # one iteration
            chunk = len(foos)//nbr_processes
            vals = list(pool.map(Foo.g, foos, chunksize=chunk))

            foos = modify_foos()

foos 具有 1000 元素。无法提前知道算法收敛的速度以及执行了多少次迭代，可能是数千次。

Answer 1

在 UNIX 变体下，进程要轻得多。 Windows 进程很繁重，需要更多时间才能启动。线程是在 Windows 上进行多处理的推荐方式。 您也可以关注此线程： Why is creating a new process more expensive on Windows than Linux?

Answer 2

您的设置对多处理并不公平。你甚至包括了不必要的primes = None 分配。 ;)

几点：

---

数据大小

您生成的数据可以让您收回创建流程的开销。尝试使用range(1_000_000) 而不是range(5000)。在将multiprocessing.start_method 设置为“spawn”（Windows 上的默认设置）的 Linux 上，这会画出不同的画面：

Concurrent test
Time: 0.957883
Number of primes: 89479

Sequential test
Time: 1.235785
Number of primes: 89479

Multiprocessing test
Time: 0.714775
Number of primes: 89479

---

重复使用您的游泳池

只要您在程序中留下了您想要稍后并行化的任何代码，就不要离开池的 with 块。如果您在开始时只创建一次池，那么将池创建包含在您的基准测试中根本没有多大意义。

---

Numpy

Numpy 部分能够释放全局解释器锁 (GIL)。这意味着，您可以从多核并行中受益，而无需创建进程的开销。无论如何，如果您正在做数学，请尝试尽可能多地使用 numpy。使用 numpy 的代码尝试 concurrent.futures.ThreadPoolExecutor 和 multiprocessing.dummy.Pool。