标签: python-multiprocessing

使用Python 3.4

我有一个失败的目标函数，并且 apply_async 调用我的错误回调函数而不是我的回调函数。

问题是捕获并发送到 error_callback 的异常没有帮助，我无法弄清楚目标函数中发生错误的位置。

我的错误回调

  def error_callback(self,e):
    print('error_callback()',e)
    # tested one of these at a time
    traceback.print_exc()
    traceback.print_last()
    traceback.print_stack()

和回溯输出

打印_执行

  File "/usr/lib/python3.4/traceback.py", line 125, in _iter_chain
    context = exc.__context__
  AttributeError: 'NoneType' object has no attribute '__context__'

最后打印

  File "/usr/lib/python3.4/traceback.py", line 262, in print_last
    raise ValueError("no last exception")
  ValueError: no last exception

打印堆栈

  File "./relative/path/to/my/script.py", line 71, in traceback
    traceback.print_stack()

我不得不承认，我对回溯库以及如何使用它不太满意，但在我看来，在这种情况下不可能使用回溯？

我应该怎么办？

我编写了 3 个不同的代码来比较有线程和没有线程。基本上测量通过使用线程节省了多少时间，结果没有任何意义。

这是我的代码：

 import time



def Function():

    global x 
    x = 0

    while x < 300000000:
        x += 1
    print x

e1 = time.clock()
E1 = time.time()

Function() 

e2 = time.clock()
E2 = time.time()

print e2 - e1
print E2 - E1 

当我运行这个时，我得到以下输出：

26.6358742929

26.6440000534

然后我编写了另一个函数，如下所示，将计数到 3 亿拆分为计数 3、1 亿：

 import time




def Function():

    global x 
    x = 0

    while x < 100000000:
        x += 1
    print x

def Function2():

    global x 
    x = 0

    while x < 100000000: …

我正在创建一个包含大约一百个子图/轴的图形，每个子图/轴都有几千个数据点。目前，我正在循环遍历每个子图并用于plt.scatter放置点。然而，这是相当慢的。是否可以使用多个 CPU 来加速绘图，通过将工作分配给每个子图一个核心或在单个子图中绘制数据点？

到目前为止，我尝试使用joblib并行进程来创建子图，但它不是在同一图中创建新的子图，而是为每个子图生成一个新图。我尝试过后端PDF、、Qt5Agg和Agg。这是我的代码的简化示例。

import matplotlib as mpl
mpl.use('PDF')
import seaborn as sns
import matplotlib.pyplot as plt
from joblib import Parallel, delayed

def plotter(name, df, ax):
    ax.scatter(df['petal_length'], df['sepal_length'])

iris = sns.load_dataset('iris')
fig, axes = plt.subplots(3,1)

Parallel(n_jobs=2)(delayed(plotter)
    (species_name, species_df, ax)
    for (species_name, species_df), ax in zip(iris.groupby('species'), axes.ravel()))

fig.savefig('test.pdf')

设置n_jobs=1有效后，所有点都将绘制在同一个图中。然而，将其增加到一以上会创建四个数字：我启动时使用一个数字plt.subplots，然后每次ax.scatter调用一个数字。

由于我将轴从第一个图形传递到plotter，所以我不确定如何/为什么创建附加图形。matplotlib 中是否有一些后备措施，如果指定的图形被另一个绘图过程“锁定”，则会导致自动创建新图形？

任何关于如何改进我当前的方法或通过替代方法实现加速的建议都将受到赞赏。

从这个问题我了解到：

当您使用多处理打开第二个进程时，会创建一个全新的 Python 实例，具有自己的全局状态。该全局状态不共享，因此子进程对全局变量所做的更改对于父进程来说是不可见的。

为了验证这种行为，我制作了一个测试脚本：

import time
import multiprocessing as mp
from multiprocessing import Pool
x = [0]  # global
def worker(c):
    if c == 1:  # wait for proc 2 to finish; is global x overwritten by now?
        time.sleep(2)
    print('enter: x =', x, 'with id', id(x), 'in proc', mp.current_process())
    x[0] = c
    print('exit: x =', x, 'with id', id(x), 'in proc', mp.current_process())
    return x[0]

pool = Pool(processes=2)
x_vals = pool.map(worker, [1, 2])
print('parent: x =', x, 'with …

如果这很明显，请原谅我，但我无法从文档中判断使用队列时的输出是否是有序的。也就是说，如果输入 是 ，[1, 2, 3]输出将是f(1), f(2), f(3)，或者它们可能是无序的？

编辑：我知道输入是在 FIFO 中处理的。但先进先出检索就一定能保证结果也按顺序返回吗？

Python v3.5、Windows 10

我正在使用多个进程并尝试捕获用户输入。input()搜索我看到的所有内容，在使用多个进程时会发生奇怪的事情。经过 8 个小时以上的尝试，我实施的任何措施都不起作用，我确信我做错了，但我一生都无法弄清楚。

以下是演示该问题的非常精简的程序。现在，当我在 PyCharm 中运行该程序时，它可以正常工作，但是当我用来pyinstaller创建单个可执行文件时，它会失败。该程序不断陷入循环，要求用户输入如下所示的内容：。

我非常确定这与 Windows 如何从我读过的内容中获取标准输入有关。我还尝试将用户输入变量作为Queue()项目传递给函数，但存在同样的问题。我读到你应该放入input()主要的 python 进程，所以我在下面这样做了if __name__ = '__main__':

from multiprocessing import Process
import time


def func_1(duration_1):
    while duration_1 >= 0:
        time.sleep(1)
        print('Duration_1: %d %s' % (duration_1, 's'))
        duration_1 -= 1


def func_2(duration_2):
    while duration_2 >= 0:
        time.sleep(1)
        print('Duration_2: %d %s' % (duration_2, 's'))
        duration_2 -= 1


if __name__ == '__main__':

    # func_1 user input
    while True:
        duration_1 = input('Enter a positive …

我开始熟悉 Python 的multiprocessing模块。以下代码按预期工作：

#outputs 0 1 2 3
from multiprocessing import Pool
def run_one(x):
    print x
    return

pool = Pool(processes=12)
for i in range(4):
    pool.apply_async(run_one, (i,))
pool.close()
pool.join() 

但是，现在，如果我围绕上述代码包装一个函数，print则不会执行语句（或至少重定向输出）：

#outputs nothing
def run():
    def run_one(x):
        print x
        return    

    pool = Pool(processes=12)
    for i in range(4):    
        pool.apply_async(run_one, (i,))
    pool.close()
    pool.join()

如果我将run_one定义移到之外run，则输出再次是预期的，当我调用时run()：

#outputs 0 1 2 3
def run_one(x):
    print x
    return

def run():    
    pool = Pool(processes=12)
    for i in range(4):       
        pool.apply_async(run_one, …

我有一个与Python Multiprocessing Pool Map非常相似的问题 ：AttributeError: Can't pickle local object

我想我明白问题出在哪里，我只是不知道如何解决它。“Pool.map”需要一个顶级函数作为输入。但我不知道如何重写这个问题：

一个简化的代码版本：

import os as os
from multiprocessing import Pool
import numpy as np
def opti_fun_data(prediction):
    def opti_fun(x):
        def error_fun(i):
            return error_fun_opti(x,prediction,i)
        try:
            pool = Pool(np.max([os.cpu_count()-1,1]))
            error = np.mean(pool.map(error_fun, range(M)))
        finally: # To make sure processes are closed in the end, even if errors happen
            pool.close()
            pool.join()
        return error
    return opti_fun

如果我跑 opti_fun_data(prediction)(x0) 我得到

Can't pickle local object 'opti_fun_data.<locals>.opti_fun.<locals>.error_fun'

我是多处理库的新手，可以伸出援助之手。对于那些对一些背景感兴趣的人：我想为一堆不同的场景/预测最小化函数“opti_fun”。计算我的错误度量/基准（“error_fun_opti”）是安静的计算密集型，因此我尝试并行化这一步。

我尝试运行一个非常简单的 Dask 程序，如下所示：

# myfile.py
from dask.distributed import Client

client = Client()

但是当我运行这个程序时，我得到了这个奇怪的错误

    An attempt has been made to start a new process before the
    current process has finished its bootstrapping phase.

    This probably means that you are not using fork to start your
    child processes and you have forgotten to use the proper idiom
    in the main module:

        if __name__ == '__main__':
            freeze_support()
            ...

    The "freeze_support()" line can be omitted if the program
    is not going to be frozen …

我希望如果我调用apply_async一个实例方法并获得它的结果，那么所做的任何更改都将保留在分叉进程中。但是，似乎每次对 apply_async 的新调用都会创建该实例的新副本。

取以下代码：

from multiprocessing.pool import Pool


class Multitest:
    def __init__(self):
        self.i = 0

    def run(self):
        with Pool(2) as pool:
            worker_jobs = []
            for j in range(10):
                job = pool.apply_async(self.process, (j,))
                worker_jobs.append(job)

            for job in worker_jobs:
                res = job.get()
                print("input", res)

    def process(self, inp):
        print("i", self.i)
        self.i += 1

        return inp

if __name__ == '__main__':
    mt = Multitest()
    mt.run()

示例输出：

i 0
i 0
i 0
i 0
i 0
input 0
i 0
i 0
i …