标签: python-multiprocessing

以下代码不打印"here".问题是什么？我在我的两台机器上测试了它(Windows 7,Ubuntu 12.10)和 http://www.compileonline.com/execute_python_online.php 它并不是"here"在所有情况下打印.

from multiprocessing import Queue, Process


def runLang(que):
    print "start"
    myDict=dict()
    for i in xrange(10000):
        myDict[i]=i
    que.put(myDict)
    print "finish"


def run(fileToAnalyze):
    que=Queue()
    processList=[]
    dicList=[]
    langs= ["chi","eng"]
    for lang in langs:
        p=Process(target=runLang,args=(que,))
        processList.append(p)
        p.start()

    for p1 in processList:
        p1.join()

    print "here"

    for _ in xrange(len(langs)):
        item=que.get()
        print item
        dicList.append(item)

if __name__=="__main__":
    processList = []
    for fileToAnalyse in ["abc.txt","def.txt"]:
        p=Process(target=run,args=(fileToAnalyse,))
        processList.append(p)
        p.start()
    for p1 in processList:
        p1.join()

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

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。有没有办法做到这一点？

docs(python 3.4)解释说spawn,"子进程只会继承运行进程对象run()方法所需的那些资源".

但哪些对象是"必要的"？我读它的方式告诉我,从内部可以到达的所有对象run()都是"必要的",包括传递args给的参数Process.__init__,加上存储在全局变量中的任何东西,以及在全局范围内定义的类,函数及其属性.但是,这是不正确的; 以下代码确认不会继承存储在全局变量中的对象:

# running under python 3.4 / Windows
# but behaves the same under Unix
import multiprocessing as mp

x = 0
class A:
    y = 0

def f():
    print(x) # 0
    print(A.y) # 0

def g(x, A):
    print(x) # 1
    print(A.y) # 0; really, not even args are inherited?

def main():
    global x
    x = 1
    A.y = 1
    p = mp.Process(target = f)
    p.start()
    q …

我试图拆分循环即

N = 1000000
for i in xrange(N):
    #do something

使用multiprocessing.Process并且它适用于N的小值.当我使用更大的N值时会出现问题.在p.join()之前或期间发生奇怪的事情并且程序没有响应.如果我把print i,而不是q.put(i)放在函数f的定义中,一切都运行良好.

我将不胜感激任何帮助.这是代码.

from multiprocessing import Process, Queue

def f(q,nMin, nMax): # function for multiprocessing
    for i in xrange(nMin,nMax):
        q.put(i)

if __name__ == '__main__':

    nEntries = 1000000

    nCpu = 10
    nEventsPerCpu = nEntries/nCpu
    processes = []

    q = Queue()

    for i in xrange(nCpu):
        processes.append( Process( target=f, args=(q,i*nEventsPerCpu,(i+1)*nEventsPerCpu) ) )

    for p in processes:
        p.start()

    for p in processes:
        p.join()

    print q.qsize()

我在Windows 7和Windows 10笔记本电脑上都运行了一些代码:

def project(filename):
    **Do Something Here**


if __name__ == '__main__':
    pool = Pool(processes=4)
    results = [pool.apply_async(project, args=(filename,)) for filename in filenamelist]
    output = [p.get() for p in results]
    print output

两台计算机都是双核/ 4线程,因此它们都应该可以很好地运行4个进程.我遇到的问题是,当我在Windows 10上运行代码时,我会运行4个python进程,但是它们使用0%的cpu并且它们不会输出任何内容,这与Windows 7个数据库不同,它将全部运行4个线程和完美的工作.

如果我不使用多处理,代码在Windows 10笔记本电脑上工作正常,所以问题必须与此相关.使用Python的多处理在Windows 10中不起作用吗？我顺便在两台机器上运行Python 2.7.

[编辑]:Windows 7 pc处理器是i5-650,Windows 10笔记本处理器是i3-2370M

[更新]:我将笔记本电脑还原到Windows 8.1,完全相同的代码按预期运行,这绝对是Windows 10的问题.

[编辑]:我用来生成的filenamelist方法如下,但这在Windows 7上运行正常.

def get_unfinished_files(indir, outdir):
    filenamelist = []
    for filename in os.listdir(indir):
        if filename not in os.listdir(outdir) and filename.endswith('.png'):
            filenamelist.append(filename)
    return filenamelist

我有一些代码使用multiprocessing.Pool来分叉工作者并并行执行任务.我正在尝试找到运行此代码的单元测试的正确方法.

注意我并不是要尝试并行测试串行代码测试用例,我知道像nose支持的软件包.

如果我编写一个测试所述并行代码的测试函数,并尝试使用nose运行测试:nosetests tests/test_function.py非并行测试正确执行,但多处理尝试fork时并行测试失败,因为main不可导入:

      File "C:\python-2.7.10.amd64\lib\multiprocessing\forking.py", line 488, in prepare
assert main_name not in sys.modules, main_name
AssertionError: __main__
    assert main_name not in sys.modules, main_name
AssertionError:     _assert main_name not in sys.modules, main_name
_main__AssertionError
: __main__

在我终止任务之前,这只会重复.如果我修改tests/test_function.py为包括以下内容,我可以成功运行测试:

if __name__ == '__main__':
    import nose
    nose.main()

然后执行 python tests\test_function.py

那么,与单元测试包集成的"正确"方法是什么(不一定是鼻子)？

Environ:Windows 7 64位上的Python 2.7.10 amd64

我还没有找到一种很好的方法来监视Python脚本的内存使用情况multiprocessing.更具体地说,我这样说:

import time

biglist = range(pow(10, 7))
time.sleep(5)

内存使用量为1.3 GB,由两者/usr/bin/time -v和top.但现在,说我这样做:

import time
from multiprocessing import Pool

def worker(x):
    biglist = range(pow(10, 7))
    time.sleep(5)
    return

Pool(5).map(worker, range(5))

现在top报告5 x 1.3 GB,这是正确的.但/usr/bin/time -v仍然报告1.3 GB,这是没有意义的.如果它正在测量父进程的消耗,那么应该说0.如果它正在测量父进程和子进程,那么它应该报告5 x 1.3 GB.为什么说1.3 GB？现在让我们尝试copy-on-write:

import time
from multiprocessing import Pool

biglist = range(pow(10, 7))

def worker(x):
    time.sleep(5)
    return

Pool(5).map(worker, range(5))

现在/usr/bin/time -v报告1.3 GB(再次),这是正确的.但top报告6 x 1.3 GB,这是不正确的.使用copy-on-write,它应该只报告1.3 GB.

如何使用multiprocessing？可靠地监视Python脚本的内存使用情况？

我正在进行python mutliprocessing模块的第一次尝试,我遇到了一些问题.我对线程模块非常熟悉,但我需要确保我正在执行的进程并行运行.

这是我正在尝试做的概述.请忽略未声明的变量/函数之类的内容,因为我无法完全粘贴代码.

import multiprocessing
import time

def wrap_func_to_run(host, args, output):
    output.append(do_something(host, args))
    return

def func_to_run(host, args):
    return do_something(host, args)

def do_work(server, client, server_args, client_args):
    server_output = func_to_run(server, server_args)
    client_output = func_to_run(client, client_args)
    #handle this output and return a result
    return result

def run_server_client(server, client, server_args, client_args, server_output, client_output):
    server_process = multiprocessing.Process(target=wrap_func_to_run, args=(server, server_args, server_output))
    server_process.start()  
    client_process = multiprocessing.Process(target=wrap_func_to_run, args=(client, client_args, client_output))
    client_process.start()
    server_process.join()
    client_process.join()
    #handle the output and return some result    

def run_in_parallel(server, client):
    #set up commands for first …

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

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

我需要帮助才能理解multiprocessing.Queue.我遇到的问题是,queue.get(...)与调用queue.put(...)和队列缓冲区(deque)相比,获得结果是非常有趣的.

这种泄漏抽象导致我调查队列的内部.它直截了当的源代码只是指向deque实现,这似乎也很简单,我不能用它来解释我所看到的行为.另外我读到Queue使用管道,但我似乎无法在源代码中找到它.

我把它归结为一个重现问题的最小例子,我在下面指定了一个可能的输出.

import threading
import multiprocessing
import queue

q = None
def enqueue(item):
    global q
    if q is None:
        q = multiprocessing.Queue()
        process = threading.Thread(target=worker, args=(q,))  # or multiprocessing.Process Doesn't matter
        process.start()
    q.put(item)
    print(f'len putted item: {len(item)}. qsize: {q.qsize()}. buffer len: {len(q._buffer)}')


def worker(local_queue):
    while True:
        try:
            while True:  # get all items
                item = local_queue.get(block=False)
                print(f'len got item: {len(item)}. qsize: {q.qsize()}. buffer len: {len(q._buffer)}')
        except queue.Empty:
            print('empty')


if …