是否可以将多个参数传递给 multiprocessing.pool？ [复制]

Posted 2023-03-12

【中文标题】是否可以将多个参数传递给 multiprocessing.pool？ [复制]

【英文标题】：Is it possible to pass multiple args to multiprocessing.pool? [duplicate]

【发布时间】：2012-10-12 00:55:41

【问题描述】：

可能重复：Python multiprocessing pool.map for multiple arguments

我想为 multiprocessing.Pool 中的子进程提供两个参数？我觉得我在上坡。是否有可能作为 2 个 args 或一个元组，或者......对于某些人来说，它似乎工作正常，正确地传递了两个文件名（输入和输出），但随后在一个变量点意外地出现了 barfs。可悲的是，它并没有真正起作用，因为输出文件都是空的——如果我直接调用它，或者单独处理，这不会发生。还有另一个复杂的因素，被调用的例程在另一个导入的模块中。将其本地化为“foo”存根模块确实可以解决问题，但它只打印参数，而不是尝试做任何实际工作。

对于顽固地拒绝学习如何使用队列来说，这可能是一个很长的路要走，但我想确认我不会在前进的道路上取得任何进展。

fixtures/txt_data/AAD.txt obj/txt_data/AAD.txt
fixtures/txt_data/ANZSMW.txt obj/txt_data/ANZSMW.txt
fixtures/txt_data/BENPA.txt obj/txt_data/BENPA.txt
fixtures/txt_data/CBAIZQ.txt obj/txt_data/CBAIZQ.txt
Traceback (most recent call last):
  File "./jobflow.py", line 60, in <module>
    main()
  File "./jobflow.py", line 57, in main
    args.func(args)
  File "./jobflow.py", line 40, in market
    pool.map(foo, market_files())
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/pool.py", line 225, in map
  return self.map_async(func, iterable, chunksize).get()
File     "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/pool.py", line 522, in get
raise self._value
TypeError: function takes exactly 1 argument (2 given)
fixtures/txt_data/CSDO.txt obj/txt_data/CSDO.txt
fixtures/txt_data/EMB.txt obj/txt_data/EMB.txt
fixtures/txt_data/GGG.txt obj/txt_data/GGG.txt
fixtures/txt_data/IDL.txt obj/txt_data/IDL.txt

这是一个错误示例。它在文件用尽之前停止。它或者抱怨它想要 2 个 args 但当我更改它以尝试传递两个 args 时只得到 1 个：

def foo(c):
    a, b, = c
    print a, b
    market2.file_main((a, b))  # does comment/uncommenting this break it only because it's in another python file?

def market(args):
    """
    read raw ticker data files and output nice, clean, more valid ticker data files
    """
    pool = multiprocessing.Pool()

    class market_files(object):
        for infile in args.infiles:
            outfile = os.path.join(args.outdir, os.path.basename(infile))
            yield (infile, outfile)

    pool.map(foo, market_files())

【参考方案1】：

哦，等等，它确实有效，但不是直接传递多个参数，而是将它们放入一个元组中。

我通过在每次迭代中生成一个新的 Process 和 p.start() 来实现它，这会产生大量的进程；-) 但确实吞下了多个参数。

从那里开始工作，我将可迭代对象简化为一个列表（现在我已经计算出一个可迭代对象可能没问题），但我认为主要是将 args 作为元组传递。一定是剪辑室地板上弄得一团糟的情况之一，无法看到有效的解决方案。

所以在控制器中我有：

    # Create a list of filenames.
    arglist = []
    for infile in args.infiles:
        outfile = os.path.join(args.outdir, os.path.basename(infile))
        arglist.append((infile, outfile))

    # Pass each process one filename to work on.
    pool = multiprocessing.Pool()
    p = pool.map(func=market2.process, iterable=arglist)

在模块中：

    def process(x):
        # Open an input file, and output file, and do work.
        infile, outfile = x
        instream = open(infile, 'rB')
        outstream = open(outfile, 'wB')
        main(instream, outstream)
        instream.close()
        outstream.close()

4 核性能（分钟）：

单线程 = 3:54 使用 subprocess = 4:52（我认为默认情况下它会阻塞） 同时使用 Process 的 squillions = 2:41（每个进程使用 1-4% cpu 饱和所有内核） 使用池 = 2:13

【讨论】：

编辑：大声笑。我认为它正在循环推动一切通过 - 所以运行真正想要的笛卡尔积。让我解决所有问题！