标签: python-multiprocessing

所以，最近，我一直在试验多处理模块。我写了这个脚本来测试它：

from multiprocessing import Process
from time import sleep

def a(x):
    sleep(x)
    print ("goo")

a = Process(target=a(3))
b = Process(target=a(5))
c = Process(target=a(8))
d = Process(target=a(10))

if __name__ == "__main__":
    a.start()
    b.start()
    c.start()
    d.start()

但是，当我尝试运行它时，它会引发此错误：

goo
Traceback (most recent call last):
  File "C:\Users\Andrew Wong\Desktop\Python\test.py", line 9, in <module>
    b = Process(target=a(5))
TypeError: 'Process' object is not callable

......我不知道发生了什么。有谁知道发生了什么，我该如何解决？

import time
from multiprocessing import Process

def loop(limit):
    for i in xrange(limit):
        pass
    print i

limit = 100000000 #100 million

start = time.time()    

for i in xrange(5):
    p = Process(target=loop, args=(limit,))
    p.start()
p.join()

end = time.time()
print end - start

我尝试运行此代码，这是我得到的输出

99999999
99999999
2.73401999474
99999999
99999999
99999999

而有时

99999999
99999999
3.72434902191
99999999
99999999
99999999
99999999
99999999

在这种情况下，循环函数被调用了 7 次而不是 5 次。为什么会有这种奇怪的行为？

我也对p.join()声明的作用感到困惑。它是同时结束任何一个进程还是所有进程？

我在通过队列在 2 个进程之间交换对象（数据帧）时遇到了一些麻烦。

第一个进程从队列中获取数据，第二个进程将数据放入队列。put-process 更快，所以 get-process 应该通过读取所有对象来清除队列。

我有奇怪的行为，因为我的代码按预期完美运行，但仅适用于数据帧中的 100 行，对于 1000 行，get-process 始终只需要 1 个对象。

import multiprocessing, time, sys
import pandas as pd

NR_ROWS = 1000
i = 0
def getDf():
    global i, NR_ROWS
    myheader = ["name", "test2", "test3"]                
    myrow1 =   [ i,  i+400, i+250]
    df = pd.DataFrame([myrow1]*NR_ROWS, columns = myheader)
    i = i+1
    return df 


def f_put(q):
    print "f_put start"        

    while(1): 
        data = getDf()                
        q.put(data)
        print "P:", data["name"].iloc[0]         
        sys.stdout.flush()                    
        time.sleep(1.55)


def f_get(q):
    print "f_get start"    

    while(1):     
        data = pd.DataFrame() …

我试图从在 python 中并行运行的进程返回结果。这是代码。

from dateutil.parser import parse
from multiprocessing.dummy import Pool
from itertools import *
import time
import random

def abcd(a):
    time.sleep(a)
    print(a)
    return a

p = Pool(3)
a = p.starmap(abcd, zip([10, 2, 15]))
p.close()
p.join()
print(a)

结果是否总是按照 zip 中数组的顺序返回？

之前我尝试使用python中的threading模块来创建多个线程。然后我了解了GIL以及它如何不允许在一台机器上利用多个 CPU 内核。所以现在我正在尝试进行多处理（我并不严格需要单独的线程）。

这是我编写的示例代码，用于查看是否正在创建不同的进程。但是从下面的输出中可以看出，我每次都得到相同的进程 ID。所以没有创建多个进程。我错过了什么？

import multiprocessing as mp
import os

def pri():
    print(os.getpid())

if __name__=='__main__':

    # Checking number of CPU cores
    print(mp.cpu_count())

    processes=[mp.Process(target=pri()) for x in range(1,4)]

    for p in processes:
        p.start()

    for p in processes:
        p.join()

输出：

4
12554
12554
12554

所以，我正在开发一个应用程序，每次启动时都必须根据哈希列表检查 ~50 GB 的数据。显然这需要并行化，我不希望应用程序在“加载...”屏幕上挂一分半钟。

我正在使用multiprocessing.Pool'smap_async来处理这个问题；主线程调用map_async(checkfiles, path_hash_pairs, callback)并提供一个回调，告诉它在发现不匹配时发出警告。

问题是……什么都没发生。使用我的任务管理器查看 Python 进程显示它们生成然后立即终止而不做任何工作。他们从不打印任何东西，当然也永远不会完成并调用回调。

这个缩小的例子也表现出同样的问题：

def printme(x):
    time.sleep(1)
    print(x)
    return x**2

if __name__ == "__main__":
    l = list(range(0,512))

    def print_result(res):
        print(res)

    with multiprocessing.Pool() as p:
        p.map_async(printme, l, callback=print_result)
    p.join()
    time.sleep(10)

运行它，然后......没有任何反应。交换map_async的map工程完全按照预期。

我只是犯了一个愚蠢的错误还是什么？

一个managers.dict（）允许共享跨越进程字典和执行线程安全的操作。在我的情况下，每个协调器进程创建共享字典，其中m元素和n工作进程从单个字典键读取和写入。

是否managers.dict()有一个单独的 dict 或mlocks 锁，一个用于其中的每个键？

当工作人员不必相互通信时，除了共享 dict 之外，还有其他方法可以m向n工作人员共享元素吗？

相关的python-manager-dict-is-very-slow-compared-to-regular-dict

我有一个joblib's的用例Parallel, delayed。我已经包含了一些在某些条件下终止工人的功能。然而，当我这样做，我随机产生JoblibWebdriverException，Multiprocessing exception，JoblibURLerror，或只error。

令我高兴的是，我在docs 中没有找到关于如何（定义？）/捕获异常的任何部分。

当我做：

    try:
       Parallel(delayed(function))
    except (JoblibWebdriverException | error | 'Multiprocessing exception'): 
    # written with separate excepts in original code
       log_errors()

我屈服，name JoblibWebdriverException not defined然后是：

---------
Sub-process traceback
---------
Multiprocessing exception:
(trace stack)

在python中使用Parallel时如何捕获未定义的joblib异常？

我有 6 个大文件，每个文件都包含一个我使用 pickle 函数保存在硬盘中的字典对象。按顺序加载所有这些大约需要 600 秒。我想同时开始加载所有这些文件以加快进程。假设它们都具有相同的大小，我希望在 100 秒内加载它们。我使用 multiprocessing 和 apply_async 分别加载它们中的每一个，但它像顺序一样运行。这是我使用的代码，它不起作用。该代码适用于其中 3 个文件，但其中 6 个文件的代码相同。我将第三个文件放在另一个硬盘中以确保 IO 不受限制。

def loadMaps():    
    start = timeit.default_timer()
    procs = []
    pool = Pool(3)
    pool.apply_async(load1(),)
    pool.apply_async(load2(),)
    pool.apply_async(load3(),)
    pool.close()
    pool.join()
    stop = timeit.default_timer()
    print('loadFiles takes in %.1f seconds' % (stop - start))

通过使用map函数multiprocessing库中当使用 2 个以上的进程时，我看不到执行时间的差异。我正在使用 4 个内核运行程序。

实际代码非常简单，计算前 4000 个斐波那契数 4 次（= 内核数量）。它在 N 个内核之间平均分配工作（例如，当使用具有 2 个进程的池时，每个进程将计算前 4000 个斐波那契数次）。整个过程是在 N = 1 到核心数量的情况下完成的。

输出，每一行的核心数量和相应的执行时间（以秒为单位）是：

1. 3,147
2. 1,72
3. 1,896
4. 1.899

有谁知道为什么在超过 2 个内核的情况下执行时间没有减少？实际代码是：

import multiprocessing
from time import time


def F(_):
    for n in range(4 * 10 ** 3):
        a, b = 0, 1
        for i in range(0, n):
            a, b = b, a + b
    return


def pool_fib():
    n_cores = multiprocessing.cpu_count()
    args = list(range(multiprocessing.cpu_count()))
    for i in range(1, n_cores …