标签: python-multiprocessing

我正在尝试在 slurm 集群上运行 python 脚本，并且我正在使用 python 的内置multiprocessing模块。

我使用了一个非常简单的设置，出于测试目的，示例是：

len(arg_list)
Out[2]: 5

threads = multiprocessing.Pool(5)
output = threads.map(func, arg_list)

Sofunc中的 5 个参数并行应用 5 次arg_list。我想知道的是如何在 slurm 中分配正确数量的 cpu/任务以使其按预期工作。这是我的 slurm 批处理脚本的相关部分的样子：

#!/bin/bash

# Runtime and memory
#SBATCH --time=90:00:00
#SBATCH --mem-per-cpu=2G

# For parallel jobs
#SBATCH --cpus-per-task=10
##SBATCH --nodes=2
#SBATCH --ntasks=1
##SBATCH --ntasks-per-node=4  

#### Your shell commands below this line ####

srun ./script_wrapper.py 'test'

如您所见，目前我有ntasks=1和cpus-per-task=10。请注意，func 的主要部分包含一个倾向于在两个内核上运行的 scipy 例程（即使用 200% 的 cpu 使用率，这就是我想要 10 个 …

我想使用 py.test 测试函数的线程安全性我的努力：

def function_to_be_tested(arg1,arg2):
   some functionality

class Test:
   def setup()
   def teardown()
   def test_conc()
       p1=Process(taget=function_to_be_tested,args(arg1,arg2,))
       p2=Process (taget=function_to_be_tested,args(arg1,arg3,))
       p1.start()
       p2.start()
       p1.join()
       p2.join()

使用 py.test 命令执行上面的文件。这显示以下错误。

ExceptionPexpect: isalive() encountered condition where "terminated" is 0, but there was no child process. Did someone else call waitpid() on our process?

你能帮我解码这个错误，并就如何做到这一点提供指导吗？

谢谢这是我正在尝试的实际代码和堆栈跟踪：

import pytest
from multiprocessing import Process
from pexpect import pxssh

def func(cls,b):
    cls.s.sendline("bteq")
    cls.s.prompt()
    print b
    #some operations inside the bteq session

class Test:
    @classmethod
    def setup_class(cls):
        cls.s=pxssh.pxssh()
        cls.s.login("IP",'Username','Pwd')
    @classmethod
    def …

我正在创建一个多处理程序来处理多个批处理，但是我的日志记录无法将批处理记录到日志文件中，只会记录 root log.info，如何设置日志记录以正确打印到日志文件？

日志只会打印这样一行“INFO:root:this is root logging”

import logging
import concurrent.futures
def process_batchs():
    batches = [i for i in range(100)]
    logging.basicConfig(filename=r'doc\test_ProcessPoolExecutor.log', filemode='w+',level=logging.DEBUG)
    logging.info('this is root logging')
    with concurrent.futures.ProcessPoolExecutor(10) as e:
        futures = []
        for batch in batches:
            future = e.submit(f, batch)
            futures.append(future)
        while True:
            dones = [future.done() for future in futures]
            if all(dones):
               results = [future.result() for future in futures]
               print results
               break
def f(batch):
    # do some thing
    logging.info('this is sub logging' + str(batch))
    return batch


if __name__ == …

我正在尝试使用 multiprocessing.pool，它似乎工作正常，除了整个脚本一次又一次循环......

我的代码 -

from multiprocessing import Pool

print "at top!"

arr = [30, 30, 30, 30]

def fib(num):
    if num == 0 or num == 1: return 1
    else: return (fib(num - 1) + fib(num - 2))

def main():
    print "start"
    pool = Pool(processes=4)
    result = [pool.apply_async(fib, args=(arr[i],)) for i in range(4)]
    pool.close()
    pool.join()
    for res in result:
        print res.get()
    print "end"


if __name__ == '__main__':
    main()


print "end of program ?"

它返回的是这个——

at top!
start
at top!
end …

编码对我来说很新，所以如果我问的是很简单还是很白痴，请不要介意我。

我写了一个 python 脚本来做一些晦涩的事情，长话短说，我试图比较来自 4 个不同文件的类似对象的一些字符串。

首先我需要获取每个文件中的对象，然后转换我感兴趣的对象部分，以我可以比较它们的格式，然后比较，最后将结果写入文件。

不幸的是，它超级慢。我知道我的瓶颈步骤在哪里（对象转换部分，function2()），我想通过一些多处理来提高我的程序的速度。

我尝试了几件事，我唯一设法开始工作的是池功能。

所以我做了一个总结我所有步骤的大函数，并尝试让它与池工作人员一起工作。

import multiprocessing as mp


def function1(name_object,input):

    fetch the objects in input
    return objects

def function2(objects) :

    transforming the objects into strings
    return strings

def function3(strings,output) :

    compare each position of the strings
    outptut.write(results)

def function_finale(name_object,input,output) :
    objects = function1(name_object,input)
    strings = function2(objects)
    results = function3(strings,output)

if __name__ == '__main__':

    pool = mp.Pool(processes=4)

    [pool.apply_async(combine, args=(object,file)) for object in file]

（不）令人惊讶的是，池工作人员在第一个子函数之后终止了该函数。我认为我可以进行并行编程，因为我的对象可以单独处理，以任何顺序异步处理，但是一旦获取对象，就必须遵循我的函数顺序。

基本上，我想在多个进程的循环中同时处理多个对象！应该可以吧？

我有一些限制，例如，我无法在字典中存储所有内容，因为我有 300k+ 个非常重的对象（实际上，我这样做是因为内存问题）。

有什么建议吗？

干杯。

如何在 Windows 中配置多处理以使用“forkserver”方法？每当我启动 IPython 控制台并输入：

import multiprocessing
multiprocessing.set_start_method('forkserver')

错误：

ValueError: cannot find context for 'forkserver'

发生。

I have a Django ORM database (mysql or sqlite), and would like to process each row with a fairly computationally intensive operation. What I have right now is something like:

entries = Entry.objects.filter(language='')
for e in entry:
    e.language = detect_language(e.text)
    e.save()

If the database was the bottleneck, I would use a transaction to speed it up. However, the detect_language function is what takes the most time. I could try to run the script multiple times in parallel, but that would …

我认为所有可以使用的任务celery也可以通过multiprocessing库来完成。

尽管这样，不知道为什么一个应用celery，而不是multiprocessing在Python程序或Web框架，如django，flask等。

在 python 程序中，使用multiprocessing.Process. 然后这个过程正在创建一个Pool以使用该map()方法为其提供一些工作。

当程序正常运行时，一切都按预期进行。但是，当它在 PyCharm 调试器中运行时，对Pool.map永远不会返回的调用和程序被锁定。

下面的简单示例演示了该问题：

1) 代码：

import multiprocessing

def inc(a):
    return a + 1;

def func():
    p = multiprocessing.Pool(2)
    print("before map")
    res = p.map(inc, [1,4])  # ==> the method hangs in debug.
    print("after call map")
    p.close()
    p.join()
    print(res)

def main():
    p = multiprocessing.Process(target=func)
    p.start()
    p.join()

if __name__ == '__main__':
    main()

2) 程序运行时按预期输出：

before map
after call map
[2, 5]

Process finished with exit code 0

3) 程序在调试器中运行时的输出 - …

我第一次尝试使用多处理。所以我想我会做一个非常简单的测试示例，它可以分解 100 个不同的数字。

from multiprocessing import Pool
from primefac import factorint
N = 10**30
L = range(N,N + 100)
pool = Pool()
pool.map(factorint, L)

这给了我错误：

Traceback (most recent call last):
  File "test.py", line 8, in <module>
    pool.map(factorint, L)
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 251, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 567, in get
    raise self._value
AssertionError: daemonic processes are not allowed to have children

我看到Python 进程池非守护进程？讨论了这个问题，但我不明白为什么它与我的简单玩具示例相关。我究竟做错了什么？