标签: python-multiprocessing

有时我需要使用没有参数的函数的多重处理。我希望我可以做类似的事情：

from multiprocessing import Pool

def f():  # no argument
    return 1

# TypeError: f() takes no arguments (1 given)
print Pool(2).map(f, range(10))

我可以做Process(target=f, args=())，但我更喜欢的语法map/ imap/ imap_unordered。有没有办法做到这一点？

我有一个json文件目录,我试图转换为dask DataFrame并将其保存到castra.有200个文件,它们之间包含O(10**7)个json记录.代码非常简单,主要遵循教程示例.

import dask.dataframe as dd
import dask.bag as db
import json
txt = db.from_filenames('part-*.json')
js = txt.map(json.loads)
df = js.to_dataframe()
cs=df.to_castra("data.castra")

我在32核机器上运行它,但代码只使用100%的一个核心.我对文档的理解是这段代码并行执行.为什么不呢？我误解了什么吗？

我正在使用Python的多处理功能在具有大约500GB RAM的机器上并行运行我的代码.要在不同的工作者之间共享一些数组,我正在创建一个Array对象:

N = 150
ndata = 10000
sigma = 3
ddim = 3

shared_data_base = multiprocessing.Array(ctypes.c_double, ndata*N*N*ddim*sigma*sigma)
shared_data = np.ctypeslib.as_array(shared_data_base.get_obj())
shared_data = shared_data.reshape(-1, N, N, ddim*sigma*sigma)

这是完美的工作sigma=1,但是对于设备的sigma=3一个硬盘驱动器缓慢填充,直到没有可用空间,然后该过程因此异常而失败:

OSError: [Errno 28] No space left on device

现在我有两个问题:

1. 为什么这段代码甚至会写入光盘？为什么不将它全部存储在内存中？
2. 我怎么解决这个问题？我可以让Python将它全部存储在RAM中而无需将其写入HDD吗？或者我可以更改写入此阵列的硬盘吗？

编辑:我在网上找到了一些建议,该数组存储在"共享内存"中.但是该/dev/shm设备有更多的可用空间,因为/dev/sda1上面的代码填充了这个空间. 这是此代码的strace日志的(相关部分).

编辑#2:我认为我找到了解决此问题的方法.通过查看源代码,我发现multiprocessing尝试在目录中创建临时文件,该文件通过使用来确定

process.current_process()._config.get('tempdir')

在脚本开头手动设置此值

from multiprocessing import process
process.current_process()._config['tempdir'] =  '/data/tmp/'

似乎正在解决这个问题.但我认为这不是解决问题的最佳方法.那么:有没有其他建议如何处理它？

我在 Python 3.6 中进行了以下进程间通信设置：

from multiprocessing.managers import BaseManager
class MyManager(BaseManager): pass
MyManager.register('get_instance', callable=lambda:my_instance)
m = MyManager(address=('', 50000), authkey=b'key')
s = m.get_server()
s.serve_forever()

为了避免阻塞较大应用程序的消息循环，我使用了一个线程来包含此设置。所以s.serve_forever()实际上是在run一个线程的函数内部运行的。

这是根据文档完成的。而且设置本身可以很好地与调用共享实例的客户经理一起使用。

然而，我发现没有办法停止这个“serve_forever”服务器。由于文档没有提到这一点，所以我检查了源代码。据说stop_event我可以set退出循环。但它不起作用，因为accepter守护进程/线程仍在运行。我无法调用shutdown服务器对象，因为我没有套接字对象c。

那么我该如何关闭这个服务器呢？

附：使用BaseManager.start()对我来说并不是一个真正的选择，因为在我的例子中，服务器共享一个只能由启动进程访问的异步消息循环。BaseManager.start()产生一个不再有权访问消息循环的新进程。get_server().serve_forever()另一方面，在调用进程内运行。

我正在使用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

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

我遇到以下情况，我在 for 循环中创建一个池，如下所示（我知道这不是很优雅，但出于酸洗原因我必须这样做）。假设pathos.multiprocessing相当于 python 的multiprocessing库（因为它涉及到一些与此问题无关的细节）。我想执行以下代码：

self.pool = pathos.multiprocessing.ProcessingPool(number_processes)


for i in range(5):


    all_responses = self.pool.map(wrapper_singlerun, range(self.no_of_restarts))

    pool._clear()

现在我的问题：循环成功运行第一次迭代。然而，在第二次迭代时，算法突然停止（没有完成操作pool.map。我怀疑生成了僵尸进程，或者该进程以某种方式存在switched。下面您将找到我迄今为止尝试过的所有内容。

for i in range(5):

    pool = pathos.multiprocessing.ProcessingPool(number_processes)

    all_responses = self.pool.map(wrapper_singlerun, range(self.no_of_restarts))

    pool._clear()

    gc.collect()

    for p in multiprocessing.active_children():
        p.terminate()
        gc.collect()

    print("We have so many active children: ", multiprocessing.active_children()) # Returns []

上面的代码在我的 Mac 上运行得很好。但是，当我将其上传到具有以下规格的集群时，我收到错误，它在第一次迭代后卡住了：

DISTRIB_ID=Ubuntu
DISTRIB_RELEASE=18.04
DISTRIB_CODENAME=bionic
DISTRIB_DESCRIPTION="Ubuntu 18.04 LTS"

这是pathos的多处理库文件的链接

multiprocessing我正在使用 SLURM 在集群上测试 Python 的模块。我想绝对确保我的每个任务实际上都按照我的意图在单独的 cpu 核心上运行。由于配置 SLURM 的可能性有很多，因此这一点并不明显。

因此，我想知道是否有一种方法可以从正在运行的Python任务中获取核心特定信息。我需要我的 Python 脚本来获取有关其运行的核心的信息，以便区分各个核心。

考虑以下脚本。已启动 10 个任务。有没有办法让它们每个人打印有关它们所分配到的核心的核心特定信息？

import multiprocessing

def hello():
    print "Hello World"

pool = multiprocessing.Pool()
jobs = []
for j in range(10):
    p = multiprocessing.Process(target = hello)
    jobs.append(p)
    p.start()

我在这里和这里找到的模块提供了 CPU 特定信息，但这些是关于可用 CPU 的信息，而不是运行时的核心信息。

在我的代码中，我使用 multiprocessing.Pool 来同时运行一些代码。简化的代码看起来有点像这样：

    class Wrapper():
        session: Session       


        def __init__(self):
            self.session = requests.Session()
            # Session initialization


        def upload_documents(docs):
            with Pool(4) as pool:
                upload_file = partial(self.upload_document)
                pool.starmap(upload_file, documents)

                summary = create_summary(documents)

            self.upload_document(summary)


        def upload_document(doc):
            self.post(doc)


        def post(data):
            self.session.post(self.url, data, other_params)

所以基本上通过 HTTP 发送文档是并行的。现在我想测试这段代码，但无法做到。这是我的测试：

    @patch.object(Session, 'post')
    def test_study_upload(self, post_mock):
        response_mock = Mock()
        post_mock.return_value = response_mock
        response_mock.ok = True

        with Wrapper() as wrapper:
            wrapper.upload_documents(documents)

        mc = post_mock.mock_calls

在调试中我可以检查模拟调用。有一个看起来有效，它是上传摘要的那个，以及一堆诸如call.json()、call.__len__()等call.__str__()调用。

没有电话上传文件。当我在方法中设置断点时upload_document，我可以看到每个文档都调用一次它，它按预期工作。但是，我无法测试它，因为我无法通过模拟验证此行为。我认为这是因为有许多进程调用同一个模拟，但仍然 - 我该如何解决这个问题？

我使用Python 3.6