标签: python-multiprocessing

从线程到进程，我已切换到并发。futures，并且希望获得/保留在各种情况下在 aThreadPoolExecutor和 a之间切换的灵活性ProcessPoolExecutor。然而，尽管承诺有一个统一的外观，但当我切换到使用时，我很难将多处理Queue对象作为参数传递：futures.submit()ProcessPoolExecutor

import multiprocessing as mp\nimport concurrent.futures\n\ndef foo(q):\n    q.put('hello')\n\nif __name__ == '__main__':\n\n    executor = concurrent.futures.ProcessPoolExecutor()\n    q = mp.Queue()\n    p = executor.submit(foo, q)\n    p.result()\n    print(q.get())\n

遇到来自多处理代码的以下异常：

\n

RuntimeError：队列对象只能通过继承在进程之间共享

\n

我认为这意味着它不喜欢接收队列作为参数，而是期望（不是在任何 OOP 意义上）在多处理分支上“继承它”，而不是将其作为参数获取。

不同之处在于，对于简单的多处理，这意味着当不通过并发.futures 为 \xe2\x80\x95 的外观使用它时，似乎没有这样的限制，因为以下代码可以无缝工作：

import multiprocessing as mp\n\ndef foo(q):\n    q.put('hello')\n\nif __name__ == '__main__':\n    q = mp.Queue()\n    p = mp.Process(target=foo, args=(q,))\n    p.start()\n    p.join()\n    print(q.get())\n

我想知道我在这个 \xe2\x80\x95 中缺少什么，如何ProcessPoolExecutor在使用并发.futures 时接受队列作为参数，就像使用 or ThreadPoolExecutormultiprocessing 时一样直接如上所示？ …

我正在通过 Gunicorn 运行 Django 实例。

在这个 Django 项目中，我有一个视图，它在后台启动一个多处理任务，每 x 毫秒获取一些数据，如果没有收到新请求，则在 x 秒后停止。

使用 runserver，我没有任何问题，我的多重处理继续按预期工作，并且我的视图从该多重处理中获取数据没有任何问题。

但是当我切换到 Gunicorn 时，似乎当我请求视图时，它没有检测到该进程已经在运行，并尝试并行启动一个新进程，返回错误的值。

这是代码。我的视图正在调用 NewReq 函数

import multiprocessing
manager=multiprocessing.Manager()
StateU3 = multiprocessing.Value('i', 0)
def NewReq(ID):
        if StateU3.value==0:
            p=multiprocessing.Process(target=BackgroundProcess)
            p.start()
        ...
        return (Values)

当 Gunicorn 调用该视图时，似乎 StateU3 值不在不同请求之间共享，就像 runserver 那样。

这是我的gunicorn.service 文件

[Unit]
Description=gunicorn daemon
After=network.target

[Service]
User=root
Group=www-data
WorkingDirectory=/home/project
ExecStart=/usr/local/bin/gunicorn --access-logfile - --workers 5 --bind         unix:/home/project/project.sock project.wsgi:application


[Install]
WantedBy=multi-user.target

您有任何线索吗？问题可能出在哪里？工人数量？线程数？

我正在尝试使用 python 通过 API 调用来抓取网站。由于要发出的请求数量很多，我正在考虑使用多线程或多处理来完成该任务。我对哪个更适合这项任务感到困惑。请您分享这方面的知识！

我使用 python 多处理的方式需要将相对较大的对象返回到主线程。我的例子在这里https://pastebin.com/xs0X9wWu（在pastebin上，因为它包含一个有点大的json blob，因为这个错误需要一个大对象）主要代码如下：

def generate_mutations(original):
    for node in original:
        yield original

def check(task):
    return task

if __name__ == "__main__":
    logger = log_to_stderr()
    logger.setLevel(logging.DEBUG)

    while True:
        with Pool(2) as pool:
            for result in pool.imap(check, generate_mutations(DATA)):
            #for result in map(check, generate_mutations(DATA)):
                print("Stopping")
                break
            print("Loop is done")
        print("Processing is done")

发生的情况是，在无限循环的几次迭代之后，池将无法终止并只是挂起（我看到“循环已完成”，但没有看到“处理已完成”）。中止（Ctrl+C）给出以下跟踪（我还在多处理调试日志中添加了）：

[DEBUG/MainProcess] created semlock with handle 139668948529152
[DEBUG/MainProcess] created semlock with handle 139668948525056
[DEBUG/MainProcess] created semlock with handle 139668948520960
[DEBUG/MainProcess] created semlock with handle 139668948516864
[DEBUG/MainProcess] created semlock with handle …

我正在具有 2 个 Tesla M60 GPU 的 Linux 机器上训练一些 PyTorch 模型，并使用此示例使用数据并行化在两个 GPU 之间分割训练https://towardsdatascience.com/how-to-scale-training-on-multiple -GPU-dae1041f49d2

然而，当代码到达调用 torch.multiprocessing.spawn 函数时，它会引发

类型错误：无法腌制“_thread.lock”对象

我知道最好的做法是发布一个小的可重现的示例，但是代码库非常大，并且我无法在足够小的模型上重现此错误以在此处发布。python 错误回溯也没有多大帮助，因为它只是引用多处理库的内部结构。我知道如果没有代码示例，这听起来可能有点太笼统了，但我只是想知道是否有人在 Linux 机器上使用 PyTorch 进行多处理时遇到了同样的错误，如果是这样，需要注意哪些事项来尝试和修理它。代码在单个 GPU 上运行得非常好，因此事实证明很难调试。

如果有帮助，我正在使用 python 3.8.11 和 torch 1.9.0，CUDA 版本 11.1

我写了一个简单的例子来说明我到底在想什么。可能有一些非常简单的解释我只是错过了。

import time
import multiprocessing as mp
import os


class SomeOtherClass:
    def __init__(self):
        self.a = 'b'


class SomeProcessor(mp.Process):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def run(self):
        soc = SomeOtherClass()
        print("PID: ", os.getpid())
        print(soc)

if __name__ == "__main__":
    queue = mp.Queue()

    for n in range(10):
        queue.put(n)

    processes = []

    for proc in range(mp.cpu_count()):
        p = SomeProcessor(queue)
        p.start()
        processes.append(p)

    for p in processes:
        p.join()

结果是：

PID: 11853
<__main__.SomeOtherClass object at 0x7fa637d3f588>
PID: 11854
<__main__.SomeOtherClass object at 0x7fa637d3f588>
PID: 11855
<__main__.SomeOtherClass object …

好吧，这有点复杂，但我有一个带有很多异步代码的异步类。

我希望在该类中并行化一个任务，并且我想生成多个进程来运行阻塞任务，并且在每个进程中我想创建一个asyncio循环来处理各种子任务。

所以我无法使用 ThreadPollExecutor 来做到这一点，但是当我尝试使用 ProcessPoolExecutor 时，我收到一个 Can't pickle local object 错误。

这是我的代码的简化版本，与 ThreadPoolExecutor 一起运行。如何与 ProcessPoolExecutor 并行？

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor


class MyClass:
    def __init__(self) -> None:
        self.event_loop = None
        # self.pool_executor = ProcessPoolExecutor(max_workers=8)
        self.pool_executor = ThreadPoolExecutor(max_workers=8)
        self.words = ["one", "two", "three", "four", "five"]
        self.multiplier = int(2)

async def subtask(self, letter: str):
    await asyncio.sleep(1)
    return letter * self.multiplier

async def task_gatherer(self, subtasks: list):
    return await asyncio.gather(*subtasks)

def blocking_task(self, word: str):
    time.sleep(1)
    subtasks …

我有一个 Python 多处理池，它的工作时间很长，即使经过彻底的调试，它也不够健壮，不会每 24 小时左右失败一次，因为它依赖于许多具有复杂交互的第三方非 Python 工具。另外，底层机器存在某些我无法控制的问题。请注意，失败并不是指整个程序崩溃，而是部分或大部分进程由于某些错误而变得空闲，并且应用程序本身挂起或仅使用未失败的进程继续工作。

我现在的解决方案是定期手动终止该作业，然后从原来的位置重新启动。

即使它并不理想，我现在想做的是：从 Python 代码本身以编程方式定期重新启动多处理池。我真的不在乎这是否意味着在工作中杀死泳池工人。哪种方法是最好的方法？

我的代码如下所示：

with Pool() as p:
    for _ in p.imap_unordered(function, data):
        save_checkpoint()
        log()

我的想法是这样的：

start = 0
end = 1000  # magic number
while start + 1 < len(data):
    current_data = data[start:end]
    with Pool() as p:
        for _ in p.imap_unordered(function, current_data):
            save_checkpoint()
            log()
            start += 1
            end += 1

或者：

start = 0
end = 1000  # magic number
while start + 1 < len(data):
    current_data = data[start:end]
    start_timeout(time=TIMEOUT) # …

我是多处理新手，我正在尝试编写一个程序来获取 google 上搜索查询的前 10 个结果。在此示例中，我只想同时运行 2 个搜索查询。这是我所拥有的：

import threading
from multiprocessing.pool import Pool
import pycountry
import bs4
import requests
from googlesearch import search

def getGoogleResults(query):
    links = []
    # from geeks4geeks
    print("Getting google results...")
    for j in search(query, tld="co.in", num=10, stop=10, pause=2):
        links.append(j)
    print("Got google results!")
    return links

global queryResults
queryResults = {}

queries = ["stackoverflow", "github"]

if __name__ == "__main__":
    with Pool(2) as p:
            p.map(getGoogleResults, queries)

但是，当我运行它时，出现以下错误：

File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/pool.py", line 771, in get
    raise self._value
multiprocessing.pool.MaybeEncodingError: Error sending result: …

multiprocessing.Value("d")返回一个类型multiprocessing.sharedctypes.Synchronized

所以目前我输入如下提示：
int_val: Synchronized = multiprocessing.Value("i")

有类似的东西Synchronized[int]吗？（类似于列表和其他容器类型）