Ошибка при использовании многопроцессорности с Cython - PullRequest
Новая
       13

Ошибка при использовании многопроцессорности с Cython

0 голосов
/ 24 ноября 2018

Я написал класс, который использует служебный модуль Cython.Затем я попытался ускорить процесс с помощью Multiprocessing для одновременной обработки нескольких экземпляров класса, но получил ошибку.Error sending result: '(0, <MemoryView of 'ndarray' at 0x19de04081f0>)'. Reason: 'TypeError('no default __reduce__ due to non-trivial __cinit__',)' Я рассмотрел написание функции __reduce__, но обо всем, что я видел, относится к выбору классов, а не методов или модулей.Я также изучал написание метода __cinit__, но видел еще меньше, что казалось уместным.

Ниже приведено упрощенное представление макетов пакетов и модулей, которые генерируют ошибку (на самом деле будут обрабатываться сотни объектов DNG, каждый из которых ссылается на уникальный файл размером 20 миллионов МБ, а ljpeg действительно содержит сотни строк ивызывается от десятков до сотен раз для каждого DNG).В этом примере ошибку можно исправить, удалив объявления типов массива, но если бы я сделал это на самом деле, то прирост производительности был бы на несколько порядков больше выигрыша от многопроцессорной обработки.

Можно ли это исправить без замедленияэто заметно или серьезный рефакторинг, и если да, то как?

sequence.py 

import multiprocessing

import numpy as np

from dng import DNG


def test_decode():
    input_file = np.zeros(3000, dtype=np.intc)

    pool = multiprocessing.Pool()
    tasks = []

    for i in range(10):
        task = pool.apply_async(thread, (i, input_file))
        tasks.append(task)

    pool.close()
    pool.join()

    for task in tasks:
        print(task.get())


def thread(i, input_file):
    dng = DNG(input_file)
    return i, dng.image


if __name__ == '__main__':
    test_decode()

dng.py 

import numpy as np

import ljpeg


class DNG:
    def __init__(self, input_file):

        self.image = ljpeg.decode(input_file)

ljpeg.pyx 

cpdef int[:] decode(int[:] encoded_image):
    encoded_image = __bar(encoded_image, 10000, 1000)
    return encoded_image


cdef int[:] __bar(int[:] array, int i, int ii):
    for j in range(i):
        for jj in range(ii):
            array = __foo(array)
    return array


cdef int[:] __foo(int[:] array):
    array[0] += 1
    return array

выход: 

Traceback (most recent call last):
  File "F:/Documents/Python/threading_multi/sequence.py", line 31, in <module>
    test_decode()
  File "F:/Documents/Python/threading_multi/sequence.py", line 22, in test_decode
    print(task.get())
  File "C:\Python36\lib\multiprocessing\pool.py", line 644, in get
    raise self._value
multiprocessing.pool.MaybeEncodingError: Error sending result: '(0, <MemoryView of 'ndarray' at 0x19de04081f0>)'. Reason: 'TypeError('no default __reduce__ due to non-trivial __cinit__',)'

Process finished with exit code 1

1 Ответ

0 голосов
/ 24 ноября 2018

Я почти уверен, что это ошибка, возвращающая просмотры памяти, сгенерированные каждым потоком, в основной поток (потому что просмотр памяти не может быть обработан).Однако само представление памяти оборачивает другой объект Python, который, вероятно, может быть засечен.

Нет необходимости указывать тип возвращаемого значения decode (или сделать его cpdef), поскольку он вызывается только из Python.В конце decode верните .base обзора памяти, чтобы получить нижележащий объект, который он переносит: 

def decode(int[:] encoding_image):
    # ...
    return encoding_image.base
...