基于map的Python多处理

Question

#!/usr/bin/env python

import multiprocessing

def tempx((the_time)):
    return int(the_time)*int(the_time)

def tempy((the_time, foobar)):
    return int(the_time)/(float(foobar))+100

def mp_handler():
    p = multiprocessing.Pool(2)
    foo = p.map(tempx, [('2')])
    print foo
    foo = p.map(tempy, [('100', '100000')])
    print foo
if __name__ == '__main__':
    mp_handler()

我有两个输入参数不同的方法。第一个方法只有：the_time，第二个方法只有the_time and foobar

我需要按特定顺序显示结果，因此我使用了map函数。但是，据我所知，上面的代码根本没有使用多进程模块，因为我使用的是两个map函数。我说得对吗？你知道吗

最终目标是同时运行两个方法。你知道吗

我错过了什么？你知道吗

达诺，这是我正在做的一件事

    import multiprocessing

def print_timing(func):
    def wrapper(*arg):
        t1 = time.time()
        res = func(*arg)
        t2 = time.time()
        print '%s took %0.3f ms' % (func.func_name, (t2-t1)*1000.0)
        return res
    return wrapper
@print_timing
def case_one(power_pred, power_core, num_thrs, possible_frequency, clamp_range):
    ds1_cur_freq = list()
    ds1_freq_index = list()
    ds1_cur_clamp = list()
    return ds1_cur_freq, ds1_freq_index, ds1_cur_clamp

@print_timing
def case_two(cpower_pred, power_core, num_thrs, possible_frequency, TT_index, DT_index, clamp_range):
    ds2_cur_freq = list()
    ds2_freq_index = list()
    ds2_cur_clamp = list()
    return ds2_cur_freq, ds2_freq_index, ds2_cur_clamp

def defs_finder():
    cpower_pred = list()
    power_pred = list()
    power_core = list()
    num_thrs = 3
    possible_frequency = list()
    clamp_range= list()
    DT_index =1
    TT_index = 0
    p = multiprocessing.Pool(2)
    #Case 1: DS1
#    ds1_cur_freq, ds1_freq_index, ds1_cur_clamp =
    ds1 = p.apply_async(case_one, args=(power_pred, power_core, num_thrs, possible_frequency))
    #Case 1: DS1
 #   ds1_cur_freq, ds1_freq_index, ds1_cur_clamp = case_one(power_pred, power_core, num_thrs, possible_frequency, clamp_range)
    #Case 2: DS2
#    ds2_cur_freq, ds2_freq_index, ds2_cur_clamp = case_two(cpower_pred, power_core, num_thrs, possible_frequency, TT_index, DT_index, clamp_range)
    ds2 = p.apply_async(case_two, args=(cpower_pred, power_core, num_thrs, possible_frequency, TT_index, DT_index))
    print ds1
    print ds2
    print ds1.get()
    print ds2.get()
#    ds1_cur_freq = ds1.get()[0]
#    ds1_freq_index = ds1.get()[1]
#    ds1_cur_clamp = ds1.get()[2]
#    ds2_cur_freq = ds2.get()[0]
#    ds2_freq_index = ds2.get()[1]
#    ds2_cur_clamp = ds2.get()[2]

defs_finder()

这就是它现在是如何实现的，这个bug是如何重现的

Answer 1

不，p.map将很好地分配您的计算：它将您传递给它的函数的求值与您传递给它的参数并行。但是，在您的代码中，p.map到tempx和tempy的应用程序不是并行的，如果您希望这样的话。你知道吗

但是，与普通的map或列表理解（[function(x) for x in your list]）不同，p.map是异步的，不会按特定顺序返回结果。之后需要对某个键上的结果进行排序。你知道吗

Answer 2

Pool.map在需要对iterable的所有元素并行运行特定函数并阻塞直到整个iterable被处理时非常有用。在您的例子中，您只是在iterable中传递一个项，所以您只是在子进程中运行一个函数，并阻塞直到它完成。这比只在父进程中运行函数要慢，因为这样会增加IPC的开销。你知道吗

如果您的目标是只使用一组参数并行运行tempx和tempy，那么^{}是一个更好的选择：

import multiprocessing

def tempx(the_time):
    return int(the_time)*int(the_time)

def tempy(the_time, foobar):
    return int(the_time)/(float(foobar))+100

def mp_handler():
    p = multiprocessing.Pool(2)
    foox = p.apply_async(tempx, args=('2',))
    fooy = p.apply_async(tempy, args=('100', '100000'))
    print foox.get()
    print fooy.get()

if __name__ == '__main__':
    mp_handler()

apply_async是非阻塞的；它立即返回一个^{}对象，您可以稍后使用它通过调用AsyncResult.get来实际获取异步操作的结果。因此，我们只需在两个函数上调用apply_async，在后台启动它们，然后在每个AsyncResult上调用get()，等待它们完成。你知道吗

这里还需要注意一点：在您的示例中，您在子进程中所做的工作非常简单—两个函数都不需要很长时间就可以完成。相比之下，生成后台进程并通过IPC将函数及其参数传递给这些后台进程，然后将结果发送回来的成本很高。您可能会发现使用multiprocessing比在父进程中按顺序执行这些函数要慢。为了使multiprocessing值得使用，您需要在tempx和tempy内部进行更昂贵的计算。你知道吗