标签: python-multiprocessing

我目前正在尝试使用带有多处理后端的 joblib 库在 python 3.5 中运行并行进程。但是，每次运行时都会出现此错误：

Process ForkServerPoolWorker-5:
Traceback (most recent call last):
  File "/opt/anaconda3/lib/python3.5/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/opt/anaconda3/lib/python3.5/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/anaconda3/lib/python3.5/multiprocessing/pool.py", line 108, in worker
    task = get()
  File "/opt/anaconda3/lib/python3.5/site-packages/joblib/pool.py", line 362, in get
    return recv()
  File "/opt/anaconda3/lib/python3.5/multiprocessing/connection.py", line 251, in recv
    return ForkingPickler.loads(buf.getbuffer())
TypeError: __new__() missing 1 required positional argument: 'path'

这是我用来运行它的 joblib 代码：

from joblib import Parallel, delayed
results = Parallel(n_jobs=6) (
             delayed(func)(i) for i in array)

默认情况下，后端是多处理的。当我将后端更改为“线程”时，代码运行良好，但对于我的用例，与使用多处理相比，线程效率低下。

我还尝试使用以下代码直接使用多处理，但我仍然遇到相同的错误： …

我正在使用multiprocessing.Pool并行化我正在运行的程序的某些部分。我正在循环数据，计算一些内容，然后返回结果。

性能不佳的代码：

def likelihood_data(self, data):
    func = partial(likelihood, means=self.means, stddevs=self.stddevs, c_ks=self.c_k)
    if len(data) > 100:
        pool = Pool(10)
        try:
            likelihoods = pool.map(func, data)
        finally:
            pool.close()
            pool.join()
    else:
        likelihoods = []
        for sample in data:
            likelihoods.append(self.likelihood(sample))
    return np.mean(likelihoods)

def likelihood(sample, means, stddevs, c_ks):  # is outside of class
    likel = [] 
    for c_k, m, s in zip(c_ks, means, stddevs):
        likel.append(likel_bound(np.log(c_k) + np.sum(logg(sample, m, s))))
    return np.sum(np.exp(likel))   

从使用 cProfile 来看，性能不佳是因为大部分时间都花在了{method 'acquire' of '_thread.lock' objects}. 我不明白为什么当每个进程彼此独立时会发生这种情况。这里发生了什么？

编辑：或者它只是花费了最长的时间，因为它正在等待所有进程完成？

我知道这听起来像是以前被问过的问题，但是等等，我会解释为什么其他选项不起作用。

我目前正在使用multiprocessing.Pool在应用程序中实现并行性，并希望扩展它以能够利用嵌套并行性。Pool仅将对象作为参数传递的天真的方法apply_async不起作用，如其他答案中所述，因为Pool无法进行腌制。

这是我的要求：

1. 我需要某种池来限制并发执行任务的数量。例如，multiprocess.Pool用于此目的，但它不能传递给其他进程。

2. 我需要嵌套并行性。在我的应用程序中，我需要执行 I/O 来识别嵌套工作是什么，所以我绝对不想从单个线程执行此操作。我认为这排除了这个问题的所有答案。

3. 它需要在标准库中；我无法添加依赖项。这就排除了这个答案。

4. 我真的很希望它能够与 Python 2 和 3 一起使用。但是，如果可以证明迁移到 Python 3 可以解决我的问题，我会考虑它。

我不需要专门使用多个进程，使用线程就可以了，因为大部分工作是 I/O 或等待子进程完成。

我尝试过使用multiprocessing.dummy，它是相同的接口，但在threading. 但是，当我尝试调用get()以检索测试结果时，我收到以下错误，所以我认为这已经结束了。

  File "/usr/lib/python2.7/multiprocessing/pool.py", line 567, in get
    raise self._value
ValueError: signal only works in main thread

我知道concurrent.futuresPython 3 中的库，但这似乎有一些严重的限制。例如，本节中的第二个示例在我的例子中似乎是一个阻碍：

https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor

我不明白如何使用基本上任何直接编写的嵌套并行算法来避免遇到这个问题。因此，即使我愿意使用 Python 3，我也认为这是不可能的。

如果没有编写我自己的实现，我不知道标准库中是否有任何其他可用选项。

我有一个非常大的目录，想要递归，将文件发送到发现并行处理的多处理队列。

PATH = "/mnt/large_directory"

def main():

#Initialize
print("Starting the scanner in root " + PATH)

with os.scandir(PATH) as it:
  for entry in it:
      if not entry.name.startswith('.') and entry.is_file():
          print(entry.name)
          print(entry.path)

我需要使用一个函数来对每个文件执行工作：

def doWork()
//Calculates md5 sum and other check sums

执行此操作的正确方法是什么，以便在工作线程可用时将文件添加到发现的多处理队列中？

我正在尝试从大约 100 个文件中拆分和重新排序数据。这些文件包含有关大约的信息。40 万用户，按时间顺序排列。目的是遍历 100 个文件并为每个用户创建一个单独的文件。

该方法是使用 python 多处理库来创建多个进程： - 一个工作进程处理池，用于加载数据、对其进行排序并将批次附加到队列中。每个批次包含一个用户的数据。- 一个从队列中取出元素并将批次添加到队列中的进程

现在，下面的代码适用于装有 Windows 10、Python 3.7 的笔记本电脑。然而，我试图在带有 ubuntu 16.04 和 python 3.5 的服务器上运行它。虽然在 Windows 上代码运行没有问题，但服务器内存不足。现在笔记本电脑有 8GB 的​​ RAM，而服务器有 256GB。我在这里缺少什么？

编辑：我已将代码替换为您应该能够运行的简化版本。代码现在创建进程，将 numpy 数组添加到队列中。另一个进程使数组出列。在数组排队期间，打印队列的大小。由于元素出列，队列大小保持较低。但是，内存使用量不断增加。

import numpy as np
import sys
import multiprocessing
import time


def queue_remover(q):

    while(1):
        if not q.empty():
            data = q.get()


def queue_adder(q, unused_number):

    for _ in range(5):
        q.put(np.zeros((1000,1000)))

    print('q size ', q.qsize())
    sys.stdout.flush()

    time.sleep(1)


if __name__ == "__main__":

    list_of_numbers = list(range(500))

    m = multiprocessing.Manager()
    queue = m.Queue(maxsize=40000)

    writer = …

我有计算数组*值的索引并返回一个字符串的程序。我使用.starmap_async()是因为我必须将两个参数传递给我的异步函数。该程序如下所示：

import multiprocessing as mp
from multiprocessing import freeze_support

def go_async(self, index, value) :
        return str(index * int(value))

def log_result(self, result):
        print("Succesfully get callback! With result: ", result)

def main() :
    array = [1,3,4,5,6,7]
    pool = mp.Pool() 
    res = pool.starmap_async(go_async, enumerate(array), callback = log_result)        
    print("Final result: ", res)
    pool.close()
    pool.join()

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

我想得到一个 str 数组的结果：

res = ['0', '3', '8', '15', '24', '35']

但我只有结果：

最终结果：位于 0x000001F7C10E51D0 的 multiprocessing.pool.MapResult 对象

如何正确地从中获取价值.starmap_async()？而且，回调不引发。

我想在所有 Pandas 列上并行应用一些函数。例如，我想并行执行此操作：

def my_sum(x, a):
    return x + a


df = pd.DataFrame({'num_legs': [2, 4, 8, 0],
                   'num_wings': [2, 0, 0, 0]})
df.apply(lambda x: my_sum(x, 2), axis=0)

我知道有一个swifter包，但它不支持axis=0应用：

NotImplementedError：Swifter 无法在大型数据集上执行 axis=0 应用。Dask 目前没有实现 axis=0 应用。更多详情请访问https://github.com/jmcarpenter2/swifter/issues/10

Dask 也不支持此功能axis=0（根据swifter 中的文档）。

我用谷歌搜索了几个来源，但找不到简单的解决方案。

不敢相信这在熊猫中如此复杂。

如何终止loop.run_in_executor与ProcessPoolExecutor优雅？启动程序后不久，发送 SIGINT (ctrl + c)。

def blocking_task():
    sleep(3)

async def main():
    exe = concurrent.futures.ProcessPoolExecutor(max_workers=4)
    loop = asyncio.get_event_loop()
    tasks = [loop.run_in_executor(exe, blocking_task) for i in range(3)]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print('ctrl + c')

随着max_workers比的任务，一切正常的数量等于或更少。但是如果max_workers更大，上面代码的输出如下：

Process ForkProcess-4:
Traceback (most recent call last):
  File "/usr/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/usr/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.8/concurrent/futures/process.py", line 233, in _process_worker
    call_item = call_queue.get(block=True)
  File "/usr/lib/python3.8/multiprocessing/queues.py", …

我有一些关于使用torch.multiprocessing模块的问题。假设我有一个torch.nn.Module电话model，我打电话model.share_memory()给它。

如果两个线程同时调用forward(), 即会发生什么情况model(input)？安全吗？或者我应该使用锁定机制来确保model多个线程不会同时访问？类似地，如果两个或多个线程有一个优化器工作model.parameters()并且它们同时调用会发生什么optimizer.step()？

我问这些问题是因为我经常看到在optimizer.step()没有锁定机制的共享模型上被调用（即在 A3C 或 ACER 的 RL 实现中），我想知道这样做是否安全。

我有一个已经可以解决的问题，multiprocessing.Pool但解决方案并不是非常理想。也就是说，我拥有一组相当小的输入，每个输入都映射到一个大数据集。虽然我可以使用imap_unordered返回列表的函数，但这远非高效，因为每个大型数据集都必须作为列表返回。

我的函数可以将它们作为生成器返回以降低延迟，但我无法从子进程返回生成器。

一个虚拟示例：

import time
import multiprocessing


def generate(x):
    for j in range(x, x + 10):
        yield j
        time.sleep(1)

def wrapper(x):
    return list(generate(x))


with multiprocessing.Pool(10) as pool:
    for ready in pool.imap_unordered(wrapper, range(0, 100, 10)):
        for item in set(ready):  # to show that order does not matter:
            print(item)

问题是，虽然整个运行现在只需要顺序运行时间的十分之一，但我仍然需要等待 10 秒才能得到第一个结果，可以通过以下方式立即获得：

def generate(x):
    for j in range(x, x + 10):
        yield j
        time.sleep(1)

for ready in map(generate, range(0, 100, 10):
    for item in set(ready):
        print(item)

它将立即打印第一项，但需要 100 …