标签: python-multiprocessing

我正在尝试导入“多处理”并使用 python 3.5.3 但它给出了错误

Traceback (most recent call last):
  File "ssser.py", line 7, in <module>
    import mutiprocessing
ImportError: No module named 'mutiprocessing'

当我尝试安装多处理模块时，我再次发生错误，我正在使用以下命令进行安装

python3 -m pip install multiprocessing

并得到错误

    Traceback (most recent call last):
      File "<string>", line 1, in <module>
      File "/tmp/pip-build-26ilgzih/multiprocessing/setup.py", line 94
        print 'Macros:'
                      ^
    SyntaxError: Missing parentheses in call to 'print'

    ----------------------------------------
Command "python setup.py egg_info" failed with error code 1 in /tmp/pip-build-26ilgzih/multiprocessing/

但是，当我在终端中导入“导入多处理”时，不会发生错误，但是当我在文件“ssscr.py”中使用时，它会给出错误，我正在树莓派 3 上使用 geany 和 python3 (IDEL)

任何人都可以帮助我如何修复这个错误吗？

我正在测试多队列。我的测试涉及一个带有 getter 和 setter 的简单类，围绕多队列。

当我重新运行 i 时，我的测试在通过和失败之间交替[编辑：当使用 block=False 时会发生这种情况。按照建议将其设置为 True 会挂起程序]。

我如何重写这个以便

(1) 清除程序执行之间的多队列以及 (2) 在 .get() 方法中读取队列中的所有值？

import multiprocessing
import queue

class MyClass:
    def __init__(self):
        self.q = multiprocessing.Queue()
        self.results = []

    def put(self, x):
        self.q.put(x)

    def get(self):
        while True:
            try:
                self.results.append(self.q.get(block=True))
            except queue.Empty:
                break
        return self.results

    @pytest.fixture
    def wrapped_queue():
        yield MyClass()

    def test_multiprocessing_queue(wrapped_queue):
        wrapped_queue.put("a")
        wrapped_queue.put("b")
        result = wrapped_queue.get()
        assert result == ["a", "b"]

我可以在同一张图中绘制单独的处理吗？例如下面的代码只能保存图中的最后一个图。

import matplotlib
matplotlib.use('agg')
from multiprocessing import Pool
import matplotlib.pyplot as plt
def f111(a,b,fig):
    plt.plot(a,b,'ro')
    plt.savefig('test.jpg')

if __name__=='__main__':
    fig=plt.figure(figsize=(20,10))
    p=Pool(8)
    for i in range(8):
        p.apply_async(f111,args=(1+i*6,6+i*6,fig))
    p.close()
    p.join()

我有一份使用该multiprocessing包并通过以下方式调用函数的工作

resultList = pool.map(myFunction, myListOfInputParameters)。

输入参数列表中的每个条目都是独立于其他条目的。

这项工作将运行几个小时。出于安全原因，我想存储定期的结果，例如每小时一次。

当作业被中止并且我想根据最后一个可用备份重新启动它时，我该如何执行此操作并能够继续处理？

我正在使用 python 的 multiprocessing.Pool 和 apply_async 来调用一堆函数。

如何判断函数是否已开始由池中的成员处理或者是否位于队列中？

例如：

import multiprocessing
import time

def func(t):
    #take some time processing
    print 'func({}) started'.format(t)
    time.sleep(t)

pool = multiprocessing.Pool()

results = [pool.apply_async(func, [t]) for t in [100]*50] #adds 50 func calls to the queue

对于每个AsyncResultin，results您可以调用ready()或get(0)来查看 func 是否完成运行。但是如何知道 func 是否 已开始但尚未完成呢？

即对于给定的 AsyncResult 对象（即给定的结果元素），有没有办法查看该函数是否已被调用或者它是否位于池的队列中？

我尝试在程序的两个部分中使用 cupy，其中之一与池并行。我设法用一个简单的例子重现它：

import cupy
import numpy as np
from multiprocessing import pool


def f(x):
    return cupy.asnumpy(2*cupy.array(x))



input = np.array([1,2,3,4])
print(cupy.asnumpy(cupy.array(input)))


print(np.array(list(map(f, input))))

p = pool.Pool(4)
output = p.map(f, input)
p.close()
p.join()
print(output)

输出如下：

[1 2 3 4]
[2 4 6 8]
Exception in thread Thread-3:
Traceback (most recent call last):
  File "/usr/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.6/threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.6/multiprocessing/pool.py", line 489, in _handle_results
    task = get()
  File "/usr/lib/python3.6/multiprocessing/connection.py", line 251, in recv …

我有一项处理器非常密集的任务，大约需要 13-20 小时才能完成，具体取决于机器。似乎是通过多处理库进行并行化的明显选择。问题是......我产生的进程越多，相同的代码变得越慢。

每次迭代的时间（即运行sparse.linalg.cg所需的时间）：

183s 1 进程

245s 2 进程

312s 3个进程

383s 4个进程

诚然，虽然 2 个进程每次迭代需要多花费 30% 多一点的时间，但它同时执行 2 个进程，因此速度仍然稍快一些。但我不认为实际的数学运算本身会更慢！这些计时器直到多处理增加的开销之后才会启动。

这是我的代码的精简版本。问题行是sparse.linalg.cg 行。（我尝试过使用 MKL 与 OpenBLAS 之类的方法，并强制它们在单个线程中运行。还尝试过手动生成进程而不是使用池。没有运气。）

def do_the_thing_partial(iteration: int, iter_size: float, outQ : multiprocessing.Queue, L: int, D: int, qP: int, elec_ind: np.ndarray, Ic: int, ubi2: int,
                 K : csc_matrix, t: np.ndarray, dip_ind_t: np.ndarray, conds: np.ndarray, hx: float, dstr: np.ndarray):
    range_start = ceil(iteration * iter_size)
    range_end = ceil((iteration + 1) * iter_size)

    for rr in range(range_start, range_end):
        # do some …

显然，如果write()调用未以某种方式同步，从多个进程写入同一文件可能会导致数据损坏。请参阅另一个问题：Python multiprocessing securewriting to a file。

然而，在尝试重现此可能的错误以进行测试时，我无法导致文件消息混淆。我想这样做是为了有效地比较有锁安全性和没有锁安全性。

不做任何事情，文件似乎受到某种程度的保护。

import multiprocessing
import random

NUM_WORKERS = 10
LINE_SIZE = 10000
NUM_LINES = 10000

def writer(i):
    line = ("%d " % i) * LINE_SIZE + "\n"
    with open("file.txt", "a") as file:
        for _ in range(NUM_LINES):
            file.write(line)

def check(file):
    for _ in range(NUM_LINES * NUM_WORKERS):
        values = next(file).strip().split()
        assert len(values) == LINE_SIZE
        assert len(set(values)) == 1

if __name__ == "__main__":
    processes = []

    for i in range(NUM_WORKERS):
        process …

有人可以解释一下下面的错误吗？

multiprocessing.pool.MaybeEncodingError：发送结果时出错：“[>]”。
原因： 'NotImplementedError("[E112] 不支持 Pickling 跨度，因为跨度只是父文档的视图，不能单独存在。腌制的跨度始终必须包含其 Doc 和 Vocab，而实际上没有与直接腌制父文档相比有缺点。因此，不要腌制 Span，而是腌制它所属的 Doc，或者使用 Span.as_doc 将 Span 转换为独立的 Doc 对象。")'

下面是我的代码：

简历解析器2.py

class Resume_parser2(object):
    inputString = ''
    skill = ['Java']
    def __init__(self, resume):
        self.__matcher = Matcher(nlp.vocab)
        self.__skills = {
            'skills'    : None
        }
        self.__text        = utils.extract_text(self.__resume, os.path.splitext(self.__resume)[1])
        self.__text        = ' '.join(self.__text.split())
        self.__nlp         = nlp(self.__text)
        self.__noun_chunks = list(self.__nlp.noun_chunks)


    def __get_basic_details(self):
        skills = utils.extract_skills(self.__nlp,self.__noun_chunks)
        self.__skills['skills'] = skills
        return

    def check_skills(self):
        skill = ['Java']
        if skill in self.__skills:
            return self.__skills

def filtered_resume(resume):
    parser = Resume_parser2(resume) …

我正在开发一个项目，其中我有一个 python 模块，它实现了迭代过程，并且一些计算是由 GPU 使用 TensorFlow 2.0 执行的。该模块在单个进程中独立使用时可以正常工作。

由于我必须使用不同的参数执行多次运行，所以我想并行化调用，但是当我从不同的进程调用模块（导入张量流）时，我得到了 的CUDA_ERROR_OUT_OF_MEMORY无限循环CUDA_ERROR_NOT_INITIALIZED，因此生成的进程永远挂起。

当然，我尝试限制 GPU 内存，如果我从不同的解释器运行两个不同的 python 脚本，它可以正常工作，但在我的情况下似乎不起作用。

特别是，如果我使用

import tensorflow as tf
gpus = tf.config.experimental.list_physical_devices('GPU')
if gpus:
    try:
       # Currently, memory growth needs to be the same across GPUs
        for gpu in gpus:
            tf.config.experimental.set_memory_growth(gpu, True)
            logical_gpus = tf.config.experimental.list_logical_devices('GPU')
            print(len(gpus), "Physical GPUs,", len(logical_gpus), "Logical GPUs")
    except RuntimeError as e:
        # Memory growth must be set before GPUs have been initialized
        print(e)

我得到 的无限循环CUDA_ERROR_NOT_INITIALIZED，而如果我使用：

physical_devices = …