标签: python-multiprocessing

我对python很新.我正在使用多处理模块读取stdin上的文本行,以某种方式转换它们并将它们写入数据库.这是我的代码片段:

batch = []
pool = multiprocessing.Pool(20)
i = 0
for i, content in enumerate(sys.stdin):
    batch.append(content)
    if len(batch) >= 10000:
        pool.apply_async(insert, args=(batch,i+1))
        batch = []
pool.apply_async(insert, args=(batch,i))
pool.close()
pool.join()

现在一切正常,直到我处理我输入我的python程序的巨大输入文件(数亿行).在某些时候,当我的数据库变慢时,我看到内存已满.

经过一番播放后,事实证明pool.apply_async以及pool.map_async永远不会阻塞,因此要处理的调用队列越来越大.

我的问题的正确方法是什么？我希望我能设置一个参数,一旦达到某个队列长度,就会阻塞pool.apply_async调用.Java中的AFAIR可以为ThreadPoolExecutor提供一个具有固定长度的BlockingQueue用于此目的.

谢谢!

我有一个Python进程,使用多处理模块生成其他5个Python进程.让我们调用父进程P0和其他P1-P5.要求是,如果我们将SIGTERM发送到P0,它应该首先关闭P1到P5然后退出自己.

问题是P1和P5正在等待信号量.因此,当我将SIGTERM发送到这些进程时,它们会调用信号处理程序并退出.但由于他们正在等待信号量,他们会抛出异常.有没有办法在退出之前捕获该异常,以便P0到P5可以正常退出？

追溯:

Traceback (most recent call last):
  File "/usr/lib64/python2.7/multiprocessing/process.py", line 258, in _bootstrap
Traceback (most recent call last):
Process Process-2:
  File "/usr/lib64/python2.7/multiprocessing/process.py", line 258, in _bootstrap
Traceback (most recent call last):
self.run()
File "/usr/lib64/python2.7/multiprocessing/process.py", line 114, in run
self._target(*self._args, **self._kwargs)
Process Process-5:
Traceback (most recent call last):
File "/usr/lib64/python2.7/multiprocessing/process.py", line 258, in _bootstrap
  self.run()
File "/usr/lib64/python2.7/multiprocessing/process.py", line 114, in run
  self._target(*self._args, **self._kwargs)
File "/opt/fireeye/scripts/mip/StaticAnalysisRunner.py", line 45, in run
  qsem.acquire()

我的代码(遗传优化算法的一部分)并行运行几个进程,等待所有进程完成,读取输出,然后用不同的输入重复.当我用60次重复测试时,一切都很好.由于它有效,我决定使用更真实的重复次数,200.我收到了这个错误:

File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_inner
 self.run()
File "/usr/lib/python2.7/threading.py", line 504, in run
 self.__target(*self.__args, **self.__kwargs)
File "/usr/lib/python2.7/multiprocessing/pool.py", line 302, in _handle_workers
 pool._maintain_pool()
File "/usr/lib/python2.7/multiprocessing/pool.py", line 206, in _maintain_pool
 self._repopulate_pool()
File "/usr/lib/python2.7/multiprocessing/pool.py", line 199, in _repopulate_pool
 w.start()
File "/usr/lib/python2.7/multiprocessing/process.py", line 130, in start
 self._popen = Popen(self)
File "/usr/lib/python2.7/multiprocessing/forking.py", line 120, in __init__
 self.pid = os.fork()
OSError: [Errno 12] Cannot allocate memory

这是我使用池的代码片段:

def RunMany(inputs):
from multiprocessing import cpu_count, Pool
proc=inputs[0]
pool=Pool(processes = proc) 
results=[]
for arg1 in inputs[1]:
    for arg2 …

我是multiprocessingpython中的新软件包,对于那些了解更多内容的人来说,我的困惑可能很容易.我一直在阅读有关并发的内容,并且已经搜索了其他类似的问题并且一无所获.(仅供参考我不希望使用multithreading,因为GIL将我的应用程序有很多慢下来.)

我在事件的框架内思考.我希望有多个进程在运行,等待事件发生.如果事件发生,它将被分配给特定进程,该进程运行然后返回其空闲状态.可能有更好的方法来做到这一点,但我的理由是我应该生成所有进程一次并使它们无限期地打开,而不是每次事件发生时创建然后关闭进程.速度对我来说是一个问题,我的事件每秒可能发生数千次.

我想出了以下玩具示例,其意图是将偶数发送到一个进程,将奇数发送到另一个进程.两个进程都是相同的,它们只是将数字附加到列表中.

from multiprocessing import Process, Queue, Pipe

slist=['even','odd']

Q={}
Q['even'] = Queue()
Q['odd'] = Queue()

ev,od = [],[]

Q['even'].put(ev)
Q['odd'].put(od)

P={}
P['even'] = Pipe()
P['odd'] = Pipe()



def add_num(s):
    """ The worker function, invoked in a process. The results are placed in
        a list that's pushed to a queue."""
#    while True :
    if not P[s][1].recv():
        print s,'- do nothing'

    else:            
        d = Q[s].get()
        print d
        d.append(P[s][1].recv())
        Q[s].put(d)
        print Q[s].get()
        P[s][0].send(False)
        print …

我正在使用python多处理功能来映射某些元素的某些功能.有点像这样:

def computeStuff(arguments, globalData, concurrent=True):
    pool = multiprocessing.Pool(initializer=initWorker, initargs=(globalData,))
    results = pool.map(workerFunction, list(enumerate(arguments)))
    return results

def initWorker(globalData):
    workerFunction.globalData = globalData

def workerFunction((index, argument)):
    ... # computation here

通常我使用cPython和Pypy在ipython中运行测试.我注意到产生的过程通常不会被杀死,所以它们开始积累,每个都使用一个ram.在计算过程中按ctrl-k会发生这种情况,这会将多处理发送到混乱的大狂热中.但即使让计算完成,这些过程也不会在Pypy中死亡.

根据文档,当池被垃圾收集时,它应该调用terminate()并终止所有进程.这里发生了什么事？我必须明确打电话close()吗？如果是,是否有某种上下文管理器可以正确管理关闭资源(即进程)？

这是在Mac OS X Yosemite上.

我已经解决了这个问题一个星期了,而且它变得非常令人沮丧,因为每次我实现一个更简单但相似的我需要做的比例示例时,事实证明多处理会捏造它.它处理共享内存的方式令我困惑,因为它非常有限,它可能会很快变得无用.

所以我的问题的基本描述是我需要创建一个进程,该进程在一些参数中传递以打开图像并创建大小为60x40的大约20K补丁.这些补丁一次保存到列表2中,需要返回到主线程,然后由GPU上运行的其他2个并发进程再次处理.

过程和工作流程以及所有大部分都需要处理的事情,我现在需要的是最容易被认为是最困难的部分.我无法保存并将带有20K补丁的列表返回到主线程.

第一个问题是因为我将这些补丁保存为PIL图像.然后我发现添加到Queue对象的所有数据都必须被pickle.第二个问题是我然后将补丁转换为每个60x40的数组并将它们保存到列表中.而现在仍然不起作用？显然,当您调用queue_obj.get()程序挂起时,队列可以保存有限数量的数据.

我尝试过很多其他的东西,而且我尝试的每一件新东西都不起作用,所以我想知道是否有人有一个库的其他建议我可以使用它来共享对象而没有所有模糊的东西？

这是我正在看的一种示例实现.请记住,这完全正常,但完全实现不会.我确实有代码打印信息性消息,以确保保存的数据具有完全相同的形状和一切,但由于某种原因它不起作用.在完整实现中,独立进程成功完成,但在q.get()处冻结.

from PIL import Image
from multiprocessing import Queue, Process
import StringIO
import numpy

img = Image.open("/path/to/image.jpg")
q = Queue()
q2 = Queue()
#
#
# MAX Individual Queue limit for 60x40 images in BW is 31,466.
# Multiple individual Queues can be filled to the max limit of 31,466.
# A single Queue can only take up to 31,466, even if split up in different puts.
def rz(patch, qn1, qn2):
    totalPatchCount = 20000
    channels …

我正在构建一个烧瓶应用程序,需要一些后台进程来运行.我决定使用multiprocess,但是在Flask中运行时它会产生两个进程.有谁知道为什么会这样？我在OS X和Ubuntu 12.04上测试了它,结果相同.这是一个例子:

import time
import multiprocessing
from flask import Flask

app = Flask(__name__)
backProc = None

def testFun():
    print('Starting')
    while True:
        time.sleep(3)
        print('looping')
        time.sleep(3)
        print('3 Seconds Later')

@app.route('/')
def root():

    return 'Started a background process with PID ' + str(backProc.pid) + " is running: " + str(backProc.is_alive())

@app.route('/kill')
def kill():
    backProc.terminate()
    return 'killed: ' + str(backProc.pid)

@app.route('/kill_all')
def kill_all():
    proc = multiprocessing.active_children()
    for p in proc:
        p.terminate()
    return 'killed all'

@app.route('/active')
def active():
    proc = multiprocessing.active_children()
    arr = [] …

简而言之

我想同时更改复杂的python对象,每个对象只由一个进程处理.我怎么能这样做(效率最高)？实施某种酸洗支持会有帮助吗？这会有效吗？

完整的问题

我有一个python数据结构ArrayDict,基本上由一个numpy数组和一个字典组成,并将任意索引映射到数组中的行.在我的例子中,所有键都是整数.

a = ArrayDict()

a[1234] = 12.5
a[10] = 3

print(a[1234])                               #12.5
print(a[10])                                 # 3.0

print(a[1234] == a.array[a.indexDict[1234]]) #true

现在我有多个这样的ArrayDicts并希望填写它们myMethod(arrayDict, params).由于myMethod价格昂贵,我想并行运行它.请注意,myMethod可能会添加许多行arrayDict.每个过程都会改变自己的过程ArrayDict.我不需要并发访问ArrayDicts.

在myMethod,我更改了条目arrayDict(即,我更改了内部numpy数组),我添加了条目arrayDict(也就是说,我向字典添加另一个索引并在内部数组中写入一个新值).最终,我希望能够在arrayDict内部numpy阵列变得太小时进行交换.这不会经常发生,如果没有更好的解决方案,我可以在程序的非并行部分执行此操作.即使没有阵列交换,我自己的尝试也没有成功.

我花了几天时间研究共享内存和python的多处理模块.由于我最终将在linux上工作,因此任务似乎相当简单:系统调用fork()允许有效地处理参数的副本.我的想法是ArrayDict在自己的进程中更改每个,返回对象的更改版本,并覆盖原始对象.为了节省内存并保存复制工作,我还使用了sharedmem数组来存储数据ArrayDict.我知道字典必须仍然被复制.

from sharedmem import sharedmem
import numpy as np

n = ...                   # length of …

所以这或多或少是一个理论问题.我有一个核心机器,据说功能强大,但只有一个核心.现在我有两个选择:

1. 多线程:就我的知识而言,即使我因为GIL而拥有它们,我也无法在我的机器中使用多个内核.因此,在这种情况下,它没有任何区别.

2. 多处理:这是我怀疑的地方.我可以在单个核心机器上进行多处理吗？或者每次我必须检查我的机器中可用的核心,然后运行完全相同或更少数量的进程？

有人可以指导一下机器中多处理和核心之间的关系.

我知道这是一个理论问题,但我的概念对此并不十分清楚.

我正在尝试了解的某些shared_memory操作。

查看源代码，看起来该模块shm_open()用于UNIX环境，并且Windows上的CreateFileMapping\ OpenFileMapping与结合使用mmap。

我从这里了解到，为了避免通过进行彻底的序列化/反序列化pickle，必须为他的共享数据类型实现__setstate__()并__getstate__()明确实现。

我在中看不到任何此类实现shared_memory.py。

如何shared_memory避开泡菜处理？

而且，在Windows机器上，仅此一项似乎在所有解释器中都可以幸免：

from mmap import mmap

shared_size = 12
shared_label = "my_mem"

mmap(-1, shared_size , shared_label)

为什么在这里需要CreateFileMapping\ OpenFileMapping？