标签: python-multiprocessing

我正在研究一个项目,使用lxml在python中同时解析多个xml文件.当我初始化进程时,我希望我的主类在将etree对象传递给进程之前对XML做一些工作,但我发现当etree对象到达新进程时,类仍然存活但是XML已经从在对象中,getroot()返回None.

我知道我只能使用队列传递可选择的数据,但是这也是我传递给'args'字段内的进程的情况吗？

这是我的代码:

import multiprocessing, multiprocessing.pool, time
from lxml import etree

def compute(tree):
    print("Start Process")
    print(type(tree)) # Returns <class 'lxml.etree._ElementTree'>
    print(id(tree)) # Returns new ID 44637320 as expected
    print(tree.getroot()) # Returns None

def pool_init(queue):
    # see http://stackoverflow.com/a/3843313/852994
    compute.queue = queue

class Main():
    def __init__(self):
        pass

    def main(self):
        tree = etree.parse('test.xml')
        print(id(tree)) # Returns object ID 43998536
        print(tree.getroot()) #Returns <Element SymCLI_ML at 0x29f5dc8>

        self.queue = multiprocessing.Queue()
        self.pool = multiprocessing.Pool(processes=1, initializer=pool_init, initargs=(self.queue,))
        self.pool.apply_async(func=compute, args=(tree,))
        time.sleep(10)

if __name__ == '__main__':
    Main().main()

任何和所有帮助非常感谢.

UPDATE/ANSWER …

我正在学习使用Python的Multiprocessing软件包来解决令人尴尬的并行问题,因此我编写了串行和并行版本来确定小于或等于自然数n的素数的数量.根据我从博客文章和Stack Overflow问题中读到的内容,我想出了以下代码:

串行

import math
import time

def is_prime(start, end):
    """determine how many primes within given range"""
    numPrime = 0
    for n in range(start, end+1):
        isPrime = True
        for i in range(2, math.floor(math.sqrt(n))+1):
            if n % i == 0:
                isPrime = False
                break
        if isPrime:
            numPrime += 1
    if start == 1:
        numPrime -= 1  # since 1 is not prime
    return numPrime

if __name__ == "__main__":
    natNum = 0
    while natNum < 2: …

我有一个像这样工作的生成器函数(Python)

def Mygenerator(x, y, z, ...):
    while True:
        # code that makes two matrices based on sequences of input arrays
        yield (matrix1, matrix2)

我想要做的是添加此生成器的输出.这条线完成了这项工作:

M1, M2 = reduce(lambda x, y: x[0] + y[0], x[1] + y[1], Mygenerator(x, y, z, ...))

我想将其并行化以加速计算.重要的是Mygenerator的输出会随着它的产生而减少,因为它list(Mygenerator(...))会占用太多内存.

我正在努力改善这段代码的执行时间.由于计算非常耗时,我认为最好的解决方案是并行化代码.

我是第一次使用这个问题中解释的地图,但后来我尝试了一种更简单的方法,认为我可以找到更好的解决方案.但是我还没有拿出任何东西,所以因为这是一个不同的问题,所以我决定把它作为一个新问题发布.

我正在使用Python 3.4在Windows平台上工作.

这是代码:

similarity_matrix = [[0 for x in range(word_count)] for x in range(word_count)]
for i in range(0, word_count):
    for j in range(0, word_count):
        if i > j:
            similarity = calculate_similarity(t_matrix[i], t_matrix[j])
            similarity_matrix[i][j] = similarity
            similarity_matrix[j][i] = similarity

这是calculate_similarity功能:

def calculate_similarity(array_word1, array_word2):
      denominator = sum([array_word1[i] + array_word2[i] for i in range(word_count)])
      if denominator == 0:
          return 0
      numerator = sum([2 * min(array_word1[i], array_word2[i]) for i in range(word_count)])
      return numerator / denominator

以及代码的解释:

• word_count 是列表中存储的唯一单词的总数 …

在我提到的这个问题中,我可以使用多处理成功运行作业,我可以看到所有处理器都在使用中.我该如何解雇这份工作？从终端我运行:

python my_multiprocessor_script.py

然后我打Ctrl+C了杀.

然而,这项工作似乎没有被杀死,我可以看到所有的处理器仍在使用中.我正在运行Red Hat Enterprise Linux Server 6.6版.

实用程序

def exec_multiprocessing(self, method, args):
    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = pool.map(method, args)
        return results

clone.py

def clone_vm(self, name, first_run, host, ip):
    # clone stuff

invoke.py

exec_args = [(name, first_run, host, ip) for host, ip in zip(hosts, ips)]
results = self.util.exec_multiprocessing(self.clone.clone_vm, exec_args)

上面的代码给出了酸洗错误。我发现这是因为我们正在传递实例方法。因此，我们应该解开实例方法。但是我无法使其工作。

注意：我无法创建顶级方法来避免这种情况。我必须使用实例方法。

添加了Python 3.4+ multiprocessing.set_start_method('spawn')，它允许创建不与主进程共享内存的进程。我正在尝试在Python 2.7中获得这种行为。是否multiprocessing.Process可以使用该类的备用端口，替代的多处理程序包或较低级别的函数？

更新：关于近距离投票，我想澄清一下，我并不是在寻求主观软件建议。相反，我正在尝试找到解决此问题的任何解决方案，并且在浏览了一段时间后，似乎可能需要使用低级功能的手动解决方案。

使用Windows和python 3.6后，一旦导入要映射的函数，就无法进行多重处理。该代码有效（给出输出[2，12，30，56，90]）：

from multiprocessing import Pool
def f(x,y):
    return x*y

if __name__ == '__main__':
    args=[(1,2),(3,4),(5,6),(7,8),(9,10)]
    with Pool(4) as p:
        result=p.starmap(f,args)
    print(result)

现在，我将函数f移到另一个名为“ test”的.py文件中，然后将其导入：

from multiprocessing import Pool
from test import f    

if __name__ == '__main__':
    args=[(1,2),(3,4),(5,6),(7,8),(9,10)]
    with Pool(4) as p:
        result=p.starmap(f,args)
    print(result)

与test.py只包含：

def f(x,y):
    return x*y

运行它会导致无限循环（不会返回任何CPU使用率很高的内容）。

是什么原因造成的，有没有办法解决？通过将所有代码复制到一个巨大的.py中，我已经成功地使多处理程序可以在一个程序上工作，这显然不理想。

我想在模型上使用多处理来使用数据框作为输入来获得预测.我有以下代码:

def perform_model_predictions(model, dataFrame, cores=4): 
    try:
        with Pool(processes=cores) as pool:
            result = pool.map(model.predict, dataFrame)
            return result
        # return model.predict(dataFrame)
    except AttributeError:
        logging.error("AttributeError occurred", exc_info=True)

我得到的错误是:

raise TypeError("sparse matrix length is ambiguous; use getnnz()"
TypeError: sparse matrix length is ambiguous; use getnnz() or shape[0]

我认为问题在于我将数据帧作为pool.map函数的第二个参数传递.任何建议或帮助将不胜感激.

I want to download and process a lot of files from website. The terms of service for the site restrict the number of files you're permitted to download per second.

The time that it takes to process the files is actually the bottle neck, so I'd like to be able process multiple files in parallel. But I don't want the different processes to combine to violate the download limit. So I need something that limits the over request rate. I …