标签: python-multiprocessing

import multiprocessing
import numpy as np
import multiprocessing as mp
import ctypes

class Test():
    def __init__(self):
        shared_array_base = multiprocessing.Array(ctypes.c_double, 100, lock=False)
        self.a = shared_array = np.ctypeslib.as_array(shared_array_base)

    def my_fun(self,i):
        self.a[i] = 1

if __name__ == "__main__":
    num_cores = multiprocessing.cpu_count()

    t = Test()

    def my_fun_wrapper(i):
        t.my_fun(i)

    with mp.Pool(num_cores) as p:
        p.map(my_fun_wrapper, np.arange(100))

    print(t.a)

在上面的代码中，我尝试编写一段代码来修改数组，使用multiprocessing. 在每个进程中执行的函数应该修改作为参数传递给索引处的my_fun()数组的值。关于上面的代码，我想知道复制的是什么。a[:]imy_fun()

1）每个进程都复制代码中的任何内容吗？我认为该物体可能是，但理想情况下什么都不是。

2）有没有办法绕过my_fun()对对象使用包装函数？

我对 python 多处理相当陌生，我正在尝试编写一个能够异步执行函数和附加回调的类。

首先，让我们为这个特定问题确定一个通用术语：

\xe2\x94\xac\xe2\x94\x80 process1 (parent)\n\xe2\x94\x94\xe2\x94\x80\xe2\x94\xac\xe2\x94\x80 process2 (child of parent1)\n  \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80\xe2\x94\x80 process3 (child of parent2)\n

在对这个主题进行了一些记录并遵循了这个SO问题之后，为了做到这一点，我为该方法提出了以下代码run：

\xe2\x94\xac\xe2\x94\x80 process1 (parent)\n\xe2\x94\x94\xe2\x94\x80\xe2\x94\xac\xe2\x94\x80 process2 (child of parent1)\n  \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80\xe2\x94\x80 process3 (child of parent2)\n

该类AsyncProcess比这个更大（它的目的是作为外部进程和新进程中的 python 函数之间的适配器multiprocessing.Process，并用于执行它们）；subprocess.Popen这就是为什么它不是它的子类multiprocessing.Process，而只是使用它（以防有人想知道）。

我在这里试图实现的是能够process3从另一个进程 () 中启动子进程 ( process2)，而无需process2等待子进程 ( process3)（因为子进程可能需要更长的时间才能启动子进程）。完成比父母）。daemon的属性没有multiprocessing.Process用，因为当父进程终止 ( process2) 时，子进程 ( process3) 也会被终止（我只想让它继续运行直到完成）。

然而，我提出的解决方案有两点我根本不喜欢：

\n
1. 我正在修改 …

我正在使用 python 的多重处理来分析一些大文本。经过几天尝试找出我的代码挂起的原因（即进程没有结束）后，我能够使用以下简单的代码重新创建问题：

import multiprocessing as mp

for y in range(65500, 65600):
    print(y)

    def func(output):

         output.put("a"*y)

    if __name__ == "__main__":

        output = mp.Queue()

        process = mp.Process(target = func, args = (output,))

        process.start()

        process.join()

正如您所看到的，如果要放入队列的项目太大，进程就会挂起。output.put()如果我在运行后编写更多代码，它不会冻结，但该过程仍然不会停止。

当字符串达到 65500 个字符时，这种情况就会开始发生，具体取决于您的解释器，它可能会有所不同。

我知道mp.Queue有一个maxsize参数，但经过一些搜索，我发现它与队列的项目数量大小有关，而不是项目本身的大小。

有没有解决的办法？我需要在原始代码中放入队列的数据非常非常大......

我想将数据缓存在torch.utils.data.Dataset. 简单的解决方案是将某些张量保留在数据集的成员中。但是，由于该类torch.utils.data.DataLoader生成多个进程，因此缓存仅位于每个实例的本地，并且可能导致我缓存相同张量的多个副本。有没有办法使用Python的多处理库在不同的加载进程之间共享数据？

当提交的任务列表长于进程数时，进程如何分配给这些任务？

from multiprocessing import Pool

def f(i):
    print(i)
    return i

with Pool(2) as pool:
    print(pool.map(f, [1, 2, 3, 4, 5]))

我正在运行一个更复杂的函数，并且执行似乎不按顺序（先进先出）。

我想使用 multiprocessing.Pool 累积总和。这是我尝试的方法：

import multiprocessing

def add_to_value(addend, value):
    value.value += addend

with multiprocessing.Manager() as manager:
    value = manager.Value(float, 0.0)
    with multiprocessing.Pool(2) as pool:
        pool.starmap(add_to_value,
                     [(float(i), value) for i in range(100)])
    print(value.value)

这会给出不正确甚至不一致的结果。例如，一次给出 2982.0，另一次给出 2927.0。正确的输出是 4950.0，当我在调用 Pool 时仅使用一个进程而不是 2 个进程时，我确实得到了这一结果。我使用的是 Python 3.7.5。

我正在尝试通过我的应用程序的多重处理进行数据库连接。我假设我调用函数的地方发生了错误。这几天我一直在为多处理而苦苦挣扎。我从https://engineering.talentpair.com/django-multiprocessing-153dbcf51dab/找到了代码

我的代码如下

视图.py

from src.conf.multiproces import MultiProcess

class GridSearchAPI(generics.GenericAPIView):
    @transaction.atomic
    def post(self, request):
        with MultiProcess() as mp:
            mp.map(request)
            returnData = mp.results()

            if returnData['result']:
                return Response(returnData, status=status.HTTP_200_OK)
            else:
                returnData["message"] = "no data found"
                return Response(result, status=status.HTTP_400_BAD_REQUEST)

多进程.py

from __future__ import print_function, absolute_import, unicode_literals
import time
try:
    from Queue import Empty
except ImportError:
    from queue import Empty

# from pathos import 
from multiprocessing import Process,cpu_count, Manager
import logging
import traceback

from django import db
from django.conf import settings
from django.core.cache …

我正在练习在 Python 中使用多处理，所以我遇到了concurrent.futuresmodule。我尝试运行以下代码：

import concurrent.futures
import time


def do_something(seconds):
    print(f'Sleeping {seconds} second(s)...')
    time.sleep(seconds)
    return f'Done Sleeping for {seconds} second(s)'

def main():

    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = [executor.submit(do_something, 1.5) for _ in range(2)]

        for f in concurrent.futures.as_completed(results):
            print(f.result())


if __name__ == '__main__':
    start = time.perf_counter()
    main()
    finish = time.perf_counter()
    print(f'Finished in {round(finish-start, 2)} second(s)')

我应该期望类似以下的输出：

Sleeping 1.5 second(s)...
Sleeping 1.5 second(s)...
Done Sleeping for 1.5 second(s)
Done Sleeping for 1.5 second(s)
Finished in 1.5 second(s)

但是，我得到了：

Done Sleeping …

我希望 Locust 使用我电脑上的所有内核。

我有很多 Locust 类，我想使用 Locust 作为库。

我的代码示例：

import gevent\nfrom locust.env import Environment\nfrom locust.stats import stats_printer\nfrom locust.log import setup_logging\nimport time\n\n\n\nfrom locust import HttpUser, TaskSet, task, between\n\n\ndef index(l):\n    l.client.get("/")\n\ndef stats(l):\n    l.client.get("/stats/requests")\n\nclass UserTasks(TaskSet):\n    # one can specify tasks like this\n    tasks = [index, stats]\n\n    # but it might be convenient to use the @task decorator\n    @task\n    def page404(self):\n        self.client.get("/does_not_exist")\n\nclass WebsiteUser(HttpUser):\n    """\n    User class that does requests to the locust web server running on localhost\n    """\n    host = "http://127.0.0.1:8089"\n …

请参阅下面的代码：

server.py

import zmq 
import time
from multiprocessing import Process
class A:
  def __init__(self):
    ctx = zmq.Context(1)
    sock = zmq.Socket(ctx, zmq.PUB)
    sock.bind('ipc://test')
    p = Process(target=A.run, args=(sock,))
    p.start()     # Process calls run, but the client can't receive messages
    p.join()      #
    #A.run(sock)  # this one is ok, messages get it to be received

  @staticmethod
  def run(sock):
    while True:
      sock.send('demo'.encode('utf-8'))
      print('sent')
      time.sleep(1)

if __name__ =='__main__':
  a = A()

client.py

import zmq 
ctx=zmq.Context(1)
sock = zmq.Socket(ctx, zmq.SUB)
sock.connect('ipc://test')
sock.setsockopt_string(zmq.SUBSCRIBE, '') 
while True:
  print(sock.recv())

在 …